future: 未來對象，或task的返回容器

1. 當(dāng)submit后：

    def submit(self, fn, *args, **kwargs):with self._shutdown_lock: # lock是線程鎖if self._shutdown:raise RuntimeError('cannot schedule new futures after shutdown')f = _base.Future() # 創(chuàng)建future對象w = _WorkItem(f, fn, args, kwargs) # 線程池執(zhí)行基本單位
self._work_queue.put(w) #實現(xiàn)的是queueself._adjust_thread_count() # 這里會進(jìn)行判斷當(dāng)前執(zhí)行線程的數(shù)量return f

?

?

2. _adjust_thread_count：

    def _adjust_thread_count(self):# When the executor gets lost, the weakref callback will wake up# the worker threads.def weakref_cb(_, q=self._work_queue):q.put(None)# TODO(bquinlan): Should avoid creating new threads if there are more# idle threads than items in the work queue.num_threads = len(self._threads)if num_threads < self._max_workers:thread_name = '%s_%d' % (self._thread_name_prefix or self,num_threads)t = threading.Thread(name=thread_name, target=_worker,args=(weakref.ref(self, weakref_cb),self._work_queue)) # 創(chuàng)建線程，并調(diào)用_worker方法，傳入work queuet.daemon = Truet.start()self._threads.add(t)_threads_queues[t] = self._work_queue

?

?

3. _worker:

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 issue16284del work_itemcontinueexecutor = 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)returndel executorexcept BaseException:_base.LOGGER.critical('Exception in worker', exc_info=True)

?

?

4. WorkItem

class _WorkItem(object):def __init__(self, future, fn, args, kwargs):self.future = futureself.fn = fnself.args = argsself.kwargs = kwargsdef run(self):if not self.future.set_running_or_notify_cancel():returntry:result = self.fn(*self.args, **self.kwargs)except BaseException as exc:self.future.set_exception(exc)# Break a reference cycle with the exception 'exc'self = Noneelse:self.future.set_result(result)

?

轉(zhuǎn)載于:https://www.cnblogs.com/callyblog/p/11147946.html

總結(jié)

以上是生活随笔為你收集整理的python threading ThreadPoolExecutor源码解析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。