python异步线程池使用

python异步线程池使用

一.昨天go协程池使用python实现

• 模拟创建10万个数字,求每个数字的2倍

二.代码

from concurrent.futures import ThreadPoolExecutor,as_completed

def execute_task(num):
    return num * 2


def main():
    tasks = 10000
    workers = 8
    fs = []
    with ThreadPoolExecutor(max_workers=workers) as executor:
        for task in range(tasks):
            f = executor.submit(execute_task,task) #返回future对象
            fs.append(f)
    results = [f.result() for f in as_completed(fs)]
    print(results)


if __name__ == "__main__":
    main()

复制

四.concurrent.futures.ThreadPoolExecutor源码

with ThreadPoolExecutor(max_workers=workers) as executor
创建了一个executer,实例了worker队列,并限制了最大的线程数,线程集合


submit(self, fn, *args, **kwargs) 向线程池队列提交异步任务,返回一个future对象,_adjust_thread_count方法则用于起新的线程,调用_worker函数,从线程池队列取出任务work_item,并执行work_item.run方法。



work_item是_WorkItem的实例 run方法将函数执行的结果通过调用future的set_result(result) 将f._state设置为了FINISHED,future的_result赋值给了result,这样后面f.result()时候可以获取结果


class _WorkItem(object):
    def __init__(self, future, fn, args, kwargs):
        self.future = future
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

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

        try:
            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 = None
        else:
            self.future.set_result(result)


复制

as_completed函数

submit函数提交任务返回的future对象,放到一个数组里,as_completed函数参数（fs, timeout=None) 调用 yield from _yield_finished_futures(finished)返回一个fs迭代器
注意:finished = [f for f in fs if f._state in [CANCELLED_AND_NOTIFIED, FINISHED]] 

再获取这些future对象的结果f.result()方法时候,由于work_item.run()已经执行完毕,f._result已经非None,变成了真正执行的函数任务结果，所以下面这行代码变通了 对futures可迭代对象遍历，获取各自的result

results = [f.result() for f in as_completed(fs)]

复制

as_completed函数

_yield_finished_futures 函数