[Python]future线程处理并发的核心源码详解

ThreadPoolExecutor 核心工作详解

1. ThreadPoolExecutor内部是如何工作的&＃xff1f;
2. 什么是future对象&＃xff1f;
3. 线程什么时候被创建&＃xff1f;
4. 任务是如何被线程处理的&＃xff1f;
5. 结束之后如何销毁的

1. ThreadPoolExecutor

我们所有的工作都是在ThreadPoolExecutor完成的&＃xff1a;

with futures.ThreadPoolExecutor(max_workers&＃61;2) as executor:...


executor中定义了几个关键的变量。

class ThreadPoolExecutor(_base.Executor):def __init__(self, max_workers&＃61;None, thread_name_prefix&＃61;&＃39;&＃39;, ...):self._max_workers &＃61; max_workers # 最大线程数self._work_queue &＃61; queue.SimpleQueue() # 任务池Queue&＃xff0c;所有线程从该Queue中取任务然后执行self._threads &＃61; set() # 线程池


2. executor.submit() 的核心

future对象是我们通过调用submit()时返回的。

future &＃61; executor.submit(sleep_and_print, _i) # submit


看看submit() 做了哪些事情&＃xff1a;

# def submit(*args, **kwargs)对参数进行了处理&＃xff0c;实际上相当于这样&＃xff1a;
def submit(self, fn, *args, **kwargs): ...# 下面是核心代码f &＃61; _base.Future() # 创建一个Future()对象w &＃61; _WorkItem(f, fn, args, kwargs) # 创建一个_WorkItem&＃xff0c;传入可调用的fn和fn的所有参数self._work_queue.put(w) # 将_WorkItem加入_work_queueself._adjust_thread_count() # 调整线程数量&＃xff0c;线程就是在这个函数中被创建的return f # 返回future


3. Future 对象

future对象初始化时&＃xff0c;定义了几个实例变量。

class Future(object):"""代表一个异步计算的结果"""def __init__(self):"""Initializes the future. Should not be called by clients."""self._condition &＃61; threading.Condition()self._state &＃61; PENDING # 状态。self._result &＃61; None # 结果。处理调用方传入的fn的结果&＃xff0c;_result &＃61; fn(*args, **kwargs)self._exception &＃61; None # 异常值。处理fn时&＃xff0c;若fn抛出异常&＃xff0c;则设置改异常值。self._done_callbacks &＃61; [] # 回调函数数组。用于处理add_done_callback传入的可调用对象。


future有五种状态&＃xff1a;

_FUTURE_STATES &＃61; [PENDING,RUNNING,CANCELLED,CANCELLED_AND_NOTIFIED,FINISHED
]


4. WorkItem 对象

WorkItem和future是一对一的&＃xff0c;submit()会为每一个future都创建一个WorkItem。WorkItem记录待处理的任务fn及其参数&＃xff0c;线程会从work_queue中取出WorkItem&＃xff0c;调用它的run()方法。WorkItem只定义了一个run()方法&＃xff1a;

class _WorkItem(object):def __init__(self, future, fn, args, kwargs):self.future &＃61; future # futureself.fn &＃61; fn # fnself.args &＃61; args # fn的参数self.kwargs &＃61; kwargs # fn的参数def run(self):if not self.future.set_running_or_notify_cancel():returntry:result &＃61; self.fn(*self.args, **self.kwargs) # 调用fnexcept BaseException as exc:self.future.set_exception(exc) # 如果fn有报错&＃xff0c;设置错误类型给future# Break a reference cycle with the exception &＃39;exc&＃39;self &＃61; Noneelse:self.future.set_result(result) # 将fn的返回值&＃xff0c;设置给future


5. executor._adjust_thread_count 函数

def _adjust_thread_count(self):def weakref_cb(_, q&＃61;self._work_queue): # 这个函数貌似没什么用&＃xff0c;后文分析...num_threads &＃61; len(self._threads) # _threads是已经创建的线程集合(set)if num_threads < self._max_workers: # 如果已创建的线程数小于最大线程数thread_name &＃61; &＃39;%s_%d&＃39; % (self._thread_name_prefix or self,num_threads) # 线程名&＃61;线程名前缀&＃43;线程序号t &＃61; threading.Thread(name&＃61;thread_name, target&＃61;_worker,args&＃61;(weakref.ref(self, weakref_cb),self._work_queue,...)) # 创建线程&＃xff0c;主要的两个参数是weakref.ref(self, weakref_cb), self._work_queue。前者是当前executor的弱引用&＃xff0c;后者是executor的任务池t.daemon &＃61; True # 守护线程。主线程退出时&＃xff0c;守护线程会自动退出t.start() # 启动线程self._threads.add(t) # 加入已创建的线程集合_threads_queues[t] &＃61; self._work_queue # _threads_queues是一个全局的弱引用WeakKeyDictionary&＃xff0c;记录所有executor的任务池&＃xff0c;以便退出时进行清理操作


t &＃61; threading.Thread(name&＃61;thread_name, target&＃61;_worker,)


这个 target 指向的 _worker 函数很关键&＃xff0c;它会执行一个 while True 循环&＃xff0c;从 work_queue 中不断获取 work_item&＃xff0c; 然后执行 work_item 的 run() 方法&＃xff1a;

def _worker(..., work_queue, ...):try:while True:work_item &＃61; work_queue.get(block&＃61;True) # 从 work_queue 中取一个 _WorkItemif work_item is not None:work_item.run() # 调用 _WorkItem.run()# del 删除对象的引用&＃xff0c;会将 x 的引用计数减一&＃xff0c;当引用计数为0时&＃xff0c;调用 x.__del__() 销毁实例。del work_itemcontinue...except BaseException:_base.LOGGER.critical(&＃39;Exception in worker&＃39;, exc_info&＃61;True)


这里没有贴出线程退出 While True 的循环&＃xff0c;是不想代码过于复杂了。如果感兴趣&＃xff0c;可以自己查看源码。

如果你对 weakref.ref(self, weakref_cb) 不理解&＃xff0c;下面我讲关键代码提炼出来&＃xff0c;模拟该弱引用&＃xff1a;

不知道什么是弱引用&＃xff0c;请先参考文章&＃xff1a;[Python] 高级用法 - 弱引用详解

import weakrefdef _worker(executor_reference):print(&＃39;Enter function: _worker&＃39;)print(executor_reference)executor &＃61; executor_reference()print(executor)if executor is not None:executor.say_hello()def run():a &＃61; A(&＃39;DemoA&＃39;)def weakref_cb(_):print(&＃39;Enter function: weakref_cb&＃39;)print((weakref.ref(a, weakref_cb)))print(weakref.ref(a))_worker((weakref.ref(a, weakref_cb))) # 两行代码效果相同_worker(weakref.ref(a)) # 两行代码效果相同class A:def __init__(self, name):self.name &＃61; namedef say_hello(self):print(f"Hello, I am {self.name}.")if __name__ &＃61;&＃61; &＃39;__main__&＃39;:run()


weakref_cb 并没有被执行&＃xff1a;

<weakref at 0x0000027068011958; to &＃39;A&＃39; at 0x000002707F073848>
<weakref at 0x0000027068011958; to &＃39;A&＃39; at 0x000002707F073848>
Enter function: _worker
<weakref at 0x0000027068011958; to &＃39;A&＃39; at 0x000002707F073848>
<__main__.A object at 0x000002707F073848>
Hello, I am DemoA.
Enter function: _worker
<weakref at 0x0000027068011958; to &＃39;A&＃39; at 0x000002707F073848>
<__main__.A object at 0x000002707F073848>
Hello, I am DemoA.


6. executor 如何退出

使用 with 关键字&＃xff0c;在结束时会调用__exit__方法&＃xff0c;这个方法调用了 shutdown() 方法。 shutdown() 会做两件事&＃xff1a;第一&＃xff0c;给 work_queue 传入一个 None 值&＃xff0c;线程执行时&＃xff0c;获取到 None 就会退出 While True 循环&＃xff1b; 第二&＃xff0c;如果 wait 为 True&＃xff0c;将剩余所有未执行完的线程&＃xff0c;调用它们的 join() 方法&＃xff0c;让主线程等待它们完成&＃xff1a;

class Executor(object):...def __enter__(self):return selfdef __exit__(self, exc_type, exc_val, exc_tb):self.shutdown(wait&＃61;True)return Falseclass ThreadPoolExecutor(_base.Executor):...def shutdown(self, wait&＃61;True):with self._shutdown_lock:self._shutdown &＃61; Trueself._work_queue.put(None)if wait:for t in self._threads:t.join()