1.//synchronous同步线程模式配置，一个链接已给线程
2.net:
3.  serviceExecutor: synchronous
4.
5.//动态线程池模式配置
6.net:
7.  serviceExecutor: adaptive  

1.//同步线程模型对应ServiceExecutorSynchronous类
2.class ServiceExecutorSynchronous final : public ServiceExecutor {
3.public:
4.    //ServiceExecutorSynchronous初始化
5.    explicit ServiceExecutorSynchronous(ServiceContext* ctx);
6.    //获取系统CPU个数
7.    Status start() override;
8.    //shutdown处理
9.    Status shutdown(Milliseconds timeout) override;
10.    //线程管理及任务入队处理
11.    Status schedule(Task task, ScheduleFlags flags) override;
12.    //同步线程模型对应mode
13.    Mode transportMode() const override {
14.        return Mode::kSynchronous;
15.    }
16.    //获取该模型统计信息
17.    void appendStats(BSONObjBuilder* bob) const override;
18.
19.private:
20.    //私有线程队列
21.    static thread_local std::deque<Task> _localWorkQueue;
22.    //递归深度
23.    static thread_local int _localRecursionDepth;
24.    //空闲线程数，例如某个链接当前没有请求，则该线程阻塞在读操作上面等待数据读到来
25.    static thread_local int64_t _localThreadIdleCounter;
26.    //shutdown的时候设置为false，链接没关闭前一直为true
27.    AtomicBool _stillRunning{false};
28.    //当前conn线程数，参考ServiceExecutorSynchronous::schedul
29.    AtomicWord<size_t> _numRunningWorkerThreads{0};
30.    //cpu个数
31.    size_t _numHardwareCores{0};
32.};  

1.//ServiceStateMachine::_scheduleNextWithGuard 启动新的conn线程
2.Status ServiceExecutorSynchronous::schedule(Task task, ScheduleFlags flags) {
3.    //如果_stillRunning为false，则直接返回
4.    if (!_stillRunning.load()) {
5.        return Status{ErrorCodes::ShutdownInProgress, "Executor is not running"};
6.    }
7.    //队列不为空，说明由任务需要运行，同步线程模型只有新连接第一次通过SSM进入该函数的时候为空
8.    //其他情况都不为空
9.    if (!_localWorkQueue.empty()) {
10.        //kMayYieldBeforeSchedule标记当返回客户端应答成功后，开始接收下一个新请求，这时候会设置该标记
11.        if (flags & ScheduleFlags::kMayYieldBeforeSchedule) {
12.            //也就是如果该链接对应的线程如果连续处理了0xf个请求，则需要休息一会儿
13.            if ((_localThreadIdleCounter++ & 0xf) == 0) {
14.                //短暂休息会儿后再处理该链接的下一个用户请求
15.                //实际上是调用TCMalloc MarkThreadTemporarilyIdle实现
16.                markThreadIdle();
17.            }
18.            //链接数即线程数超过了CPU个数，则每处理完一个请求，就yield一次
19.            if (_numRunningWorkerThreads.loadRelaxed() > _numHardwareCores) {
20.                stdx::this_thread::yield();//线程本次不参与CPU调度，也就是放慢脚步
21.            }
22.        }
23.        //带kMayRecurse标识，说明即将调度执行的是dealTask
24.        //如果递归深度小于synchronousServiceExecutorRecursionLimit，则执行task
25.        if ((flags & ScheduleFlags::kMayRecurse) &&
26.            (_localRecursionDepth < synchronousServiceExecutorRecursionLimit.loadRelaxed())) {
27.            ++_localRecursionDepth;
28.            //递归深度没有超限，则直接执行task，不用入队
29.            task();
30.        } else {
31.            //入队,等待
32.            _localWorkQueue.emplace_back(std::move(task));
33.        }
34.        return Status::OK();
35.    }
36.    //创建conn线程，线程名conn-xx(实际上是从listener线程继承过来的，这时候的Listener线程是父线程，在
37.    //ServiceStateMachine::start中已通过线程守护ThreadGuard改为conn-xx)，执行对应的task
38.    Status status = launchServiceWorkerThread([ this, task = std::move(task) ] {
39.        //说明来了一个新链接，线程数自增
40.        int ret = _numRunningWorkerThreads.addAndFetch(1);
41.        //新链接到来的第一个任务实际上是readTask任务
42.        _localWorkQueue.emplace_back(std::move(task));
43.        while (!_localWorkQueue.empty() && _stillRunning.loadRelaxed()) {
44.             //每次任务如果是通过线程私有队列获取运行，则恢复递归深度为初始值1
45.            _localRecursionDepth = 1;
46.            //取出该线程拥有的私有队列上的第一个任务运行
47.            _localWorkQueue.front()();
48.           //该任务已经执行完毕，把该任务从队列移除
49.            _localWorkQueue.pop_front();
50.        }
51.        //走到这里说明线程异常了或者需要退出，如链接关闭，需要消耗线程
52.        ......
53.    });
54.    return status;
55.}  

1.class ServiceExecutorAdaptive : public ServiceExecutor {
2.public:
3.    //初始化构造
4.    explicit ServiceExecutorAdaptive(...);
5.    explicit ServiceExecutorAdaptive(...);
6.    ServiceExecutorAdaptive(...) = default;
7.    ServiceExecutorAdaptive& operator=(ServiceExecutorAdaptive&&) = default;
8.    virtual ~ServiceExecutorAdaptive();
9.    //控制线程及worker线程初始化创建
10.    Status start() final;
11.    //shutdown处理
12.    Status shutdown(Milliseconds timeout) final;
13.    //任务调度运行
14.    Status schedule(Task task, ScheduleFlags flags) final;
15.    //adaptive动态线程模型对应Mode
16.    Mode transportMode() const final {
17.        return Mode::kAsynchronous;
18.    }
19.    //统计信息
20.    void appendStats(BSONObjBuilder* bob) const final;
21.    //获取runing状态
22.    int threadsRunning() {
23.        return _threadsRunning.load();
24.    }
25.    //新键一个worker线程
26.    void _startWorkerThread();
27.    //worker工作线程主循环while{}处理
28.    void _workerThreadRoutine(int threadId, ThreadList::iterator it);
29.    //control控制线程主循环，主要用于控制什么时候增加线程
30.    void _controllerThreadRoutine();
31.    //判断队列中的任务数和可用线程数大小，避免任务task饥饿
32.    bool _isStarved() const;
33.    //asio网络库io上下文
34.    std::shared_ptr<asio::io_context> _ioContext; //早期ASIO中叫io_service
35.    //TransportLayerManager::createWithConfig赋值调用
36.    std::unique_ptr<Options> _config;
37.    //线程列表及其对应的锁
38.    mutable stdx::mutex _threadsMutex;
39.    ThreadList _threads;
40.    //控制线程
41.    stdx::thread _controllerThread;
42.
43.    //TransportLayerManager::createWithConfig赋值调用
44.    //时间嘀嗒处理
45.    TickSource* const _tickSource;
46.    //运行状态
47.    AtomicWord<bool> _isRunning{false};
48.    //kThreadsRunning代表已经执行过task的线程总数，也就是这些线程不是刚刚创建起来的
49.    AtomicWord<int> _threadsRunning{0};
50.    //代表当前刚创建或者正在启动的线程总数，也就是创建起来还没有执行task的线程数
51.    AtomicWord<int> _threadsPending{0};
52.    //当前正在执行task的线程
53.    AtomicWord<int> _threadsInUse{0};
54.    //当前入队还没执行的task数
55.    AtomicWord<int> _tasksQueued{0};
56.    //当前入队还没执行的deferredTask数
57.    AtomicWord<int> _deferredTasksQueued{0};
58.    //TransportLayerManager::createWithConfig赋值调用
59.    //没什么实际作用
60.    TickTimer _lastScheduleTimer;
61.    //记录这个退出的线程生命期内执行任务的总时间
62.    AtomicWord<TickSource::Tick> _pastThreadsSpentExecuting{0};
63.    //记录这个退出的线程生命期内运行的总时间(包括等待IO及运行IO任务的时间)
64.    AtomicWord<TickSource::Tick> _pastThreadsSpentRunning{0};
65.    //完成线程级的统计
66.    static thread_local ThreadState* _localThreadState;
67.
68.    //总的入队任务数
69.    AtomicWord<int64_t> _totalQueued{0};
70.    //总执行的任务数
71.    AtomicWord<int64_t> _totalExecuted{0};
72.    //从任务被调度入队，到真正被执行这段过程的时间，也就是等待被调度的时间
73.    AtomicWord<TickSource::Tick> _totalSpentQueued{0};
74.
75.    //shutdown的时候等待线程消耗的条件变量
76.    stdx::condition_variable _deathCondition;
77.    //条件变量，如果发现工作线程压力大，为了避免task饥饿
78.    //通知controler线程,通知见ServiceExecutorAdaptive::schedule，等待见_controllerThreadRoutine
79.    stdx::condition_variable _scheduleCondition;
80.};  

1.//线程运行时间统计，包含两种类型时间统计
2.enum class ThreadTimer
3.{
4.    //线程执行task任务的时间+等待数据的时间
5.    Running,
6.    //只包含线程执行task任务的时间
7.    Executing
8.};
9.
10.//线程私有统计信息，记录该线程运行时间，运行时间分为两种:
11.//1. 执行task任务的时间 2. 如果没有客户端请求，线程就会等待，这就是线程等待时间
12.struct ThreadState {
13.    //构造初始化
14.    ThreadState(TickSource* ts) : running(ts), executing(ts) {}
15.    //线程一次循环处理的时间，包括IO等待和执行对应网络事件对应task的时间
16.    CumulativeTickTimer running;
17.    //线程一次循环处理中执行task任务的时间，也就是真正工作的时间
18.    CumulativeTickTimer executing;
19.    //递归深度
20.    int recursionDepth = 0;
21.};
22.
23.//获取指定which类型的工作线程相关运行时间，
24.//例如Running代表线程总运行时间(等待数据+任务处理)
25.//Executing只包含执行task任务的时间
26.TickSource::Tick ServiceExecutorAdaptive::_getThreadTimerTotal(ThreadTimer which) const {
27.    //获取一个时间嘀嗒tick
28.    TickSource::Tick accumulator;
29.    //先把已消耗的线程的数据统计出来
30.    switch (which) {
31.    //获取生命周期已经结束的线程执行任务的总时间(只包括执行任务的时间)
32.    case ThreadTimer::Running:
33.          accumulator = _pastThreadsSpentRunning.load();
34.          break;
35.     //获取生命周期已经结束的线程整个生命周期时间(包括空闲时间+执行任务时间)
36.     case ThreadTimer::Executing:
37.          accumulator = _pastThreadsSpentExecuting.load();
38.          break;
39.     }
40.     //然后再把统计当前正在运行的worker线程的不同类型的统计时间统计出来
41.     stdx::lock_guard<stdx::mutex> lk(_threadsMutex);
42.     for (auto& thread : _threads) {
43.        switch (which) {
44.            //获取当前线程池中所有工作线程执行任务时间
45.            case ThreadTimer::Running:
46.                accumulator += thread.running.totalTime();
47.                break;
48.           //获取当前线程池中所有工作线程整个生命周期时间(包括空闲时间+执行任务时间)
49.            case ThreadTimer::Executing:
50.                accumulator += thread.executing.totalTime();
51.                break;
52.        }
53.    }
54.    //返回的时间计算包含了已销毁的线程和当前正在运行的线程的相关统计
55.    return accumulator;
56.}  

1.Status ServiceExecutorAdaptive::start() {
2.    invariant(!_isRunning.load());
3.    //running状态
4.    _isRunning.store(true);
5.    //控制线程初始化创建，线程回调ServiceExecutorAdaptive::_controllerThreadRoutine
6.    _controllerThread = stdx::thread(&ServiceExecutorAdaptive::_controllerThreadRoutine, this);
7.    //启动时候默认启用CPU核心数/2个worker线程
8.    for (auto i = 0; i < _config->reservedThreads(); i++) {
9.        //创建一个工作线程
10.        _startWorkerThread();
11.    }
12.    return Status::OK();
13.}  

1.Status ServiceExecutorAdaptive::schedule(ServiceExecutorAdaptive::Task task, ScheduleFlags flags) {
2.    //获取当前时间
3.    auto scheduleTime = _tickSource->getTicks();
4.    //kTasksQueued: 普通tak，也就是dealTask
5.    //_deferredTasksQueued: deferred task,也就是readTask
6.    //defered task和普通task分开记录   _totalQueued=_deferredTasksQueued+_tasksQueued
7.    auto pendingCounterPtr = (flags & kDeferredTask) ? &_deferredTasksQueued : &_tasksQueued;
8.    //相应队列
9.    pendingCounterPtr->addAndFetch(1);
10.    ......
11.    //这里面的task()执行后-task()执行前的时间才是CPU真正工作的时间
12.    auto wrappedTask = [ this, task = std::move(task), scheduleTime, pendingCounterPtr ] {
13.        //worker线程回调会执行该wrappedTask，
14.        pendingCounterPtr->subtractAndFetch(1);
15.        auto start = _tickSource->getTicks();
16.        //从任务被调度入队，到真正被执行这段过程的时间，也就是等待被调度的时间
17.        //从任务被调度入队，到真正被执行这段过程的时间
18.        _totalSpentQueued.addAndFetch(start - scheduleTime);
19.        //recursionDepth=0说明开始进入调度处理，后续有可能是递归执行
20.        if (_localThreadState->recursionDepth++ == 0) {
21.            //记录wrappedTask被worker线程调度执行的起始时间
22.            _localThreadState->executing.markRunning();
23.            //当前正在执行wrappedTask的线程加1
24.            _threadsInUse.addAndFetch(1);
25.        }
26.        //ServiceExecutorAdaptive::_workerThreadRoutine执行wrappedTask后会调用guard这里的func
27.        const auto guard = MakeGuard([this, start] { //改函数在task()运行后执行
28.            //每执行一个任务完成，则递归深度自减
29.            if (--_localThreadState->recursionDepth == 0) {
30.                 //wrappedTask任务被执行消耗的总时间
31.                //_localThreadState->executing.markStopped()代表任务该task执行的时间
32.                _localThreadState->executingCurRun += _localThreadState->executing.markStopped();
33.                //下面的task()执行完后，正在执行task的线程-1
34.                 _threadsInUse.subtractAndFetch(1);
35.            }
36.            //总执行的任务数，task每执行一次增加一次
37.            _totalExecuted.addAndFetch(1);
38.        });
39.        //运行任务
40.        task();
41.    };
42.    //kMayRecurse标识的任务，会进行递归调用   dealTask进入调度的时候调由该标识
43.    if ((flags & kMayRecurse) && //递归调用，任务还是由本线程处理
44.        //递归深度还没达到上限，则还是由本线程继续调度执行wrappedTask任务
45.        (_localThreadState->recursionDepth + 1 < _config->recursionLimit())) {
46.        //本线程立马直接执行wrappedTask任务，不用入队到boost-asio全局队列等待调度执行
47.        //io_context::dispatch   io_context::dispatch
48.        _ioContext->dispatch(std::move(wrappedTask));
49.    } else { //入队   io_context::post
50.        //task入队到schedule得全局队列，等待工作线程调度
51.        _ioContext->post(std::move(wrappedTask));
52.    }
53.    //
54.    _lastScheduleTimer.reset();
55.    //总的入队任务数
56.    _totalQueued.addAndFetch(1);
57.    //kDeferredTask真正生效在这里
58.    //如果队列中的任务数大于可用线程数，说明worker压力过大，需要创建新的worker线程
59.    if (_isStarved() && !(flags & kDeferredTask)) {//kDeferredTask真正生效在这里
60.        //条件变量，通知controler线程,通知_controllerThreadRoutine控制线程处理
61.        _scheduleCondition.notify_one();
62.    }
63.    return Status::OK();
64.}  

1.//创建线程的回掉函数，线程循环主体，从队列获取task任务执行
2.void ServiceExecutorAdaptive::_workerThreadRoutine(
3.    int threadId, ServiceExecutorAdaptive::ThreadList::iterator state) {
4.    //设置线程模
5.    _localThreadState = &(*state);
6.    {
7.    //worker-N线程名
8.        std::string threadName = str::stream() << "worker-" << threadId;
9.        setThreadName(threadName);
10.    }
11.    //该线程第一次执行while中的任务的时候为ture，后面都是false
12.    //表示该线程是刚创建起来的，还没有执行任何一个task
13.    bool stillPending = true;
14.
15.    //线程退出的时候执行以下{}，都是一些计数清理
16.    const auto guard = MakeGuard([this, &stillPending, state] {
17.        //该worker线程退出前的计数清理、信号通知处理
18.        //......
19.    }
20.    while (_isRunning.load()) {
21.        ......
22.        //本次循环执行task的时间，不包括网络IO等待时间
23.        state->executingCurRun = 0;
24.        try {
25.            //通过_ioContext和入队的任务联系起来
26.            asio::io_context::work work(*_ioContext);
27.            //记录开始时间，也就是任务执行开始时间
28.            state->running.markRunning();
29.            //执行ServiceExecutorAdaptive::schedule中对应的task
30.            //线程第一次运行task任务,最多从队列拿一个任务执行
31.            //runTime.toSystemDuration()指定一次run最多运行多长时间
32.        if (stillPending) {
33.            //执行一个任务就会返回
34.            _ioContext->run_one_for(runTime.toSystemDuration());
35.         } else {  // Otherwise, just run for the full run period
36.                //_ioContext对应的所有任务都执行完或者toSystemDuration超时后才会返回
37.                _ioContext->run_for(runTime.toSystemDuration()); //io_context::run_for
38.         }
39.            ......
40.        }
41.        //该线程第一次执行while中的任务后设置ture，后面都是false
42.        if (stillPending) {
43.            _threadsPending.subtractAndFetch(1);
44.            stillPending = false;
45.        //当前线程数比初始线程数多
46.        } else if (_threadsRunning.load() > _config->reservedThreads()) {
47.            //代表本次循环该线程真正处理任务的时间与本次循环总时间(总时间包括IO等待和IO任务处理时间)
48.            double executingToRunning = state->executingCurRun / static_cast<double>(spentRunning);
49.            executingToRunning *= 100;
50.            dassert(executingToRunning <= 100);
51.
52.            int pctExecuting = static_cast<int>(executingToRunning);
53.            //线程很多，超过了指定配置，并且满足这个条件，该worker线程会退出，线程比较空闲，退出
54.            //如果线程真正处理任务执行时间占比小于该值，则说明本线程比较空闲，可以退出。
55.            if (pctExecuting < _config->idlePctThreshold()) {
56.                log() << "Thread was only executing tasks " << pctExecuting << "% over the last "
57.                      << runTime << ". Exiting thread.";
58.                break;  //退出线程循环，也就是线程自动销毁了
59.            }
60.        }
61.    }
62.}  

1.//controller控制线程
2.void ServiceExecutorAdaptive::_controllerThreadRoutine() {
3.    //控制线程线程名设置
4.    setThreadName("worker-controller"_sd);
5.    ......
6.    //控制线程主循环
7.    while (_isRunning.load()) {
8.        //一次while结束的时候执行对应func ,也就是结束的时候计算为起始时间
9.        const auto timerResetGuard =
10.            MakeGuard([&sinceLastControlRound] { sinceLastControlRound.reset(); });
11.         //等待工作线程通知,最多等待stuckThreadTimeout
12.        _scheduleCondition.wait_for(fakeLk, _config->stuckThreadTimeout().toSystemDuration());
13.        ......
14.        double utilizationPct;
15.        {
16.            //获取所有线程执行任务的总时间
17.            auto spentExecuting = _getThreadTimerTotal(ThreadTimer::Executing);
18.            //获取所有线程整个生命周期时间(包括空闲时间+执行任务时间+创建线程的时间)
19.            auto spentRunning = _getThreadTimerTotal(ThreadTimer::Running);
20.            //也就是while中执行一次这个过程中spentExecuting差值，
21.            //也就是spentExecuting代表while一次循环的Executing time开始值,
22.            //lastSpentExecuting代表一次循环对应的结束time值
23.            auto diffExecuting = spentExecuting - lastSpentExecuting;
24.            //也就是spentRunning代表while一次循环的running time开始值,
25.            //lastSpentRunning代表一次循环对应的结束time值
26.            auto diffRunning = spentRunning - lastSpentRunning;
27.            if (spentRunning == 0 || diffRunning == 0)
28.                utilizationPct = 0.0;
29.            else {
30.                lastSpentExecuting = spentExecuting;
31.                lastSpentRunning = spentRunning;
32.
33.                 //一次while循环过程中所有线程执行任务的时间和线程运行总时间的比值
34.                utilizationPct = diffExecuting / static_cast<double>(diffRunning);
35.                utilizationPct *= 100;
36.            }
37.        }
38.        //也就是本while()执行一次的时间差值，也就是上次走到这里的时间和本次走到这里的时间差值大于该阀值
39.        //也就是控制线程太久没有判断线程池是否够用了
40.        if (sinceLastControlRound.sinceStart() >= _config->stuckThreadTimeout()) {
41.            //use中的线程数=线程池中总的线程数，说明线程池中线程太忙了
42.            if ((_threadsInUse.load() == _threadsRunning.load()) &&
43.                (sinceLastSchedule >= _config->stuckThreadTimeout())) {
44.                log() << "Detected blocked worker threads, "
45.                      << "starting new reserve threads to unblock service executor";
46.                 //一次批量创建这么多线程，如果我们配置adaptiveServiceExecutorReservedThreads非常大
47.                //这里实际上有个问题，则这里会一次性创建非常多的线程，可能反而会成为系统瓶颈
48.                //建议mongodb官方这里最好做一下上限限制
49.                for (int i = 0; i < _config->reservedThreads(); i++) {
50.                   //创建新的worker工作线程
51.                    _startWorkerThread();
52.                }
53.            }
54.            continue;
55.        }
56.         //当前的worker线程数
57.        auto threadsRunning = _threadsRunning.load();
58.        //保证线程池中worker线程数最少都要reservedThreads个
59.        if (threadsRunning < _config->reservedThreads()) {
60.            //线程池中线程数最少数量不能比最低配置少
61.            while (_threadsRunning.load() < _config->reservedThreads()) {
62.                _startWorkerThread();
63.            }
64.        }
65.         //worker线程非空闲占比小于该阀值，说明压力不大，不需要增加worker线程数
66.        if (utilizationPct < _config->idlePctThreshold()) {
67.            continue;
68.        }
69.        //走到这里，说明整体worker工作压力还是很大的
70.        //我们在这里循环stuckThreadTimeout毫秒，直到我们等待worker线程创建起来并正常运行task
71.        //因为如果有正在创建的worker线程，我们等待一小会，最多等待stuckThreadTimeout ms
72.         //保证一次while循环时间为stuckThreadTimeout
73.        do {
74.            stdx::this_thread::sleep_for(_config->maxQueueLatency().toSystemDuration());
75.        } while ((_threadsPending.load() > 0) &&
76.                 (sinceLastControlRound.sinceStart() < _config->stuckThreadTimeout()));
77.        //队列中任务数多余可用空闲线程数，说明压力有点大，给线程池增加一个新的线程
78.        if (_isStarved()) {
79.            _startWorkerThread();
80.        }
81.    }
82.}  

while {  #等待工作线程唤醒条件变量，最长等待stuckThreadTimeout  _scheduleCondition.wait_for(stuckThreadTimeout)  ......  #获取线程池中所有线程最近一次运行任务的总有效时间TT1  Executing = _getThreadTimerTotal(ThreadTimer::Executing);  #获取线程池中所有线程最近一次运行任务的总运行时间TT2  Running = _getThreadTimerTotal(ThreadTimer::Running);  #线程池中所有线程的总有效工作时间占比 = TT1/TT2  utilizationPct = Executing / Running;  ......  #代表control线程太久没有进行线程池压力检查了  if(本次循环到该行代码的时间 > stuckThreadTimeout阀值) {  #说明太久没做压力检查，造成工作线程不够用了  if(_threadsInUse == _threadsRunning) {  #批量创建一批工作线程  for(; i < reservedThreads; i++)  #创建工作线程  _startWorkerThread();  }  #control线程继续下一次循环压力检查  continue;  }     ......  #如果当前线程池中总线程数小于最小线程数配置  #则创建一批线程，保证最少工作线程数达到要求  if (threadsRunning < reservedThreads) {  while (_threadsRunning < reservedThreads) {  _startWorkerThread();  }  }  ......  #检查上一次循环到本次循环这段时间范围内线程池中线程的工作压力  #如果压力不大，则说明无需增加工作线程数，则继续下一次循环  if (utilizationPct < idlePctThreshold) {  continue;  }  ......  #如果发现已经有线程创建起来了，但是这些线程还没有运行任务  #这说明当前可用线程数可能足够了，我们休息sleep_for会儿在判断一下  #该循环最多持续stuckThreadTimeout时间  do {  stdx::this_thread::sleep_for();  } while ((_threadsPending.load() > 0) &&  (sinceLastControlRound.sinceStart() < stuckThreadTimeout)  #如果tasksQueued队列中的任务数大于工作线程数，说明任务在排队了  #该扩容线程池中线程了  if (_isStarved()) {  _startWorkerThread();  }
}