源码分析-----ThreadPoolExecutor

创建一个线程池：

import time

from concurrent.futures import ProcessPoolExecutor, as_completed

def get_html(n):

    time.sleep(n)

    print('subprogram {} success'.format(n))

    return n

if __name__ == '__main__':

   executor = ProcessPoolExecutor(max_workers=2)

   all_task = [executor.submit(get_html, i) for i in range(3)] # 注意创建submit函数的时候，里面的参数是单个的，并不是元组形式传入。

   for futrue in as_completed(all_task):

       print(futrue.result())

subprogram 0 success

0

subprogram 1 success

1

subprogram 2 success

2

Future对象的理解：

　　在我们创建submit()的时候会立即返回一个future对象，这个对象可能是没有完成的，但是会在将来某个时候完成，所以它是一个未来对象，或者叫做task的返回容器，保存task的执行状态及执行结果。（注意future与task的区别）

看submit()源码：

f = _base.Future()

w = _WorkItem(f, fn, args, kwargs)

self._pending_work_items[self._queue_count] = w

self._work_ids.put(self._queue_count)

self._queue_count += 1

# Wake up queue management thread

self._result_queue.put(None)

self._start_queue_management_thread()

return f

首先在调用submit()的时候就创建一个future，然后将future作为参数，和线程执行函数及参数一起传入一个_WorkItem类，所以实际上_WorkItem是线程执行的一个单元，最后再将_WorkItem类放入一线程池的队列_work_ids中。然后计算加入的_WorkItem数量。self._queue_count，当启动的线程数量小于线程池的max_workers时候，就会立刻启动一个线程，该线程以_work_ids作为参数，从里面取出一个_WorkItem开始运行。

巴特西

源码分析-----ThreadPoolExecutor

最新文章

热门文章