摘要：對(duì)線程池的研究是之前對(duì)分析的附加工作。在之前對(duì)源碼分析的文章中，寫到調(diào)度器將任務(wù)放入線程池的函數(shù)這里分析的線程池類是，也就是上述代碼中所使用的類。

對(duì)Python線程池的研究是之前對(duì)Apshceduler分析的附加工作。

在之前對(duì)Apshceduler源碼分析的文章中，寫到調(diào)度器將任務(wù)放入線程池的函數(shù)

    def _do_submit_job(self, job, run_times):
        def callback(f):
            exc, tb = (f.exception_info() if hasattr(f, "exception_info") else
                       (f.exception(), getattr(f.exception(), "__traceback__", None)))
            if exc:
                self._run_job_error(job.id, exc, tb)
            else:
                self._run_job_success(job.id, f.result())

        f = self._pool.submit(_run_job, job, job._jobstore_alias, run_times, self._logger.name)
        f.add_done_callback(callback)

這里分析的線程池類是concurrent.futures.ThreadPoolExecutor，也就是上述代碼中self._pool所使用的類。先上self._pool.submit函數(shù)的代碼，再做詳細(xì)分析

    def submit(self, fn, *args, **kwargs):
        with self._shutdown_lock:
            if self._shutdown:
                raise RuntimeError("cannot schedule new futures after shutdown")

            f = _base.Future()
            w = _WorkItem(f, fn, args, kwargs)

            self._work_queue.put(w)
            self._adjust_thread_count()
            return f

f和w是兩個(gè)非常重要的變量，f作為submit返回的對(duì)象，submit函數(shù)的調(diào)用者可以對(duì)其添加回調(diào)，待fn執(zhí)行完成后，會(huì)在當(dāng)前線程執(zhí)行，具體是如何實(shí)現(xiàn)的，這里先不說(shuō)，下面再詳細(xì)分析；w則是封裝了線程需要執(zhí)行的方法和參數(shù)，通過(guò)self._work_queue.put(w)方法放入一個(gè)隊(duì)列當(dāng)中。

self._adjust_thread_count()方法則是檢查當(dāng)前線程池的線程數(shù)量，如果小于設(shè)定的最大值，就開辟一個(gè)線程，代碼就不上了，直接看這些個(gè)線程都是干嘛的

def _worker(executor_reference, work_queue):
    try:
        while True:
            work_item = work_queue.get(block=True)
            if work_item is not None:
                work_item.run()
                # Delete references to object. See issue16284
                del work_item
                continue
            executor = executor_reference()
            # Exit if:
            #   - The interpreter is shutting down OR
            #   - The executor that owns the worker has been collected OR
            #   - The executor that owns the worker has been shutdown.
            if _shutdown or executor is None or executor._shutdown:
                # Notice other workers
                work_queue.put(None)
                return
            del executor
    except BaseException:
        _base.LOGGER.critical("Exception in worker", exc_info=True)

這些線程就是一個(gè)死循環(huán)，不斷的從任務(wù)隊(duì)列中獲取到_WorkItem，然后通過(guò)其封裝方法，執(zhí)行我們需要的任務(wù)。如果取到的任務(wù)為None，就往隊(duì)列中再放入一個(gè)None，以通知其它線程結(jié)束，然后結(jié)束當(dāng)前循環(huán)。

    def run(self):
        if not self.future.set_running_or_notify_cancel():
            return

        try:
            result = self.fn(*self.args, **self.kwargs)
        except BaseException as e:
            self.future.set_exception(e)
        else:
            self.future.set_result(result)

如果沒有異常，執(zhí)行結(jié)束后，會(huì)執(zhí)行之前我們說(shuō)的回調(diào)。在self.future.set_result(result)方法中會(huì)執(zhí)行任務(wù)回調(diào)，當(dāng)然了，是在當(dāng)前線程中。如果需要寫入數(shù)據(jù)庫(kù)之類的操作，不建議在回調(diào)中直接寫入。