Python3标准库:concurrent.futures管理并发任务池
1. concurrent.futures管理并发任务池
concurrent.futures模块提供了使用工作线程或进程池运行任务的接口。线程和进程池的api是一样的,所以应用只做最小的修改就可以在线程和进程之间顺利地切换。
这个模块提供了两种类型的类与这些池交互。执行器(executor)用来管理工作线程或进程池,future用来管理工作线程或进程计算的结果。要使用一个工作线程或进程池,应用要创建适当的执行器类的一个实例,然后向它提交任务来运行。每个任务启动时,会返回一个future实例。需要任务的结果时,应用可以使用future阻塞,直到得到结果。目前已经提供了不同的api,可以很方便地等待任务完成,所以不需要直接管理future对象。
1.1 利用基本线程池使用map()
threadpoolexecutor管理一组工作线程,当这些线程可用于完成更多工作时,可以向它们传入任务。下面的例子使用map()并发地从一个输入迭代器生成一组结果。这个任务使用time.sleep()暂停不同的时间,从而展示不论任务的执行顺序如何,map()总是根据输入按顺序返回值。
from concurrent import futures import threading import time def task(n): print('{}: sleeping {}'.format( threading.current_thread().name, n) ) time.sleep(n / 10) print('{}: done with {}'.format( threading.current_thread().name, n) ) return n / 10 ex = futures.threadpoolexecutor(max_workers=2) print('main: starting') results = ex.map(task, range(5, 0, -1)) print('main: unprocessed results {}'.format(results)) print('main: waiting for real results') real_results = list(results) print('main: results: {}'.format(real_results))
map()的返回值实际上是一种特殊类型的迭代器,它知道主程序迭代处理时要等待各个响应。
1.2 调度单个任务
除了使用map(),还可以借助submit()利用一个执行器调度单个任务。然后可以使用返回的future实例等待这个任务的结果。
from concurrent import futures import threading import time def task(n): print('{}: sleeping {}'.format( threading.current_thread().name, n) ) time.sleep(n / 10) print('{}: done with {}'.format( threading.current_thread().name, n) ) return n / 10 ex = futures.threadpoolexecutor(max_workers=2) print('main: starting') f = ex.submit(task, 5) print('main: future: {}'.format(f)) print('main: waiting for results') result = f.result() print('main: result: {}'.format(result)) print('main: future after result: {}'.format(f))
任务完成之后,future的状态会改变,并得到结果。
1.3 按任意顺序等待任务
调用future的result()方法会阻塞,直到任务完成(可能返回一个值,也可能抛出一个异常)或者撤销。可以使用map()按调度任务的顺序访问多个任务的结果。如果处理结果的顺序不重要,则可以使用as_completed()在每个任务完成时处理它的结果。
from concurrent import futures import random import time def task(n): time.sleep(random.random()) return (n, n / 10) ex = futures.threadpoolexecutor(max_workers=5) print('main: starting') wait_for = [ ex.submit(task, i) for i in range(5, 0, -1) ] for f in futures.as_completed(wait_for): print('main: result: {}'.format(f.result()))
因为池中的工作线程与任务同样多,故而所有任务都可以启动。它们会按随机的顺序完成,所以每次运行这个示例程序时as_completed()生成的值都不同。
1.4 回调
要在任务完成时采取某个动作,不用显式地等待结果,可以使用add_done_callback()指示future完成时要调用一个新函数。这个回调应当是有一个参数(future实例)的callable函数。
from concurrent import futures import time def task(n): print('{}: sleeping'.format(n)) time.sleep(0.5) print('{}: done'.format(n)) return n / 10 def done(fn): if fn.cancelled(): print('{}: canceled'.format(fn.arg)) elif fn.done(): error = fn.exception() if error: print('{}: error returned: {}'.format( fn.arg, error)) else: result = fn.result() print('{}: value returned: {}'.format( fn.arg, result)) if __name__ == '__main__': ex = futures.threadpoolexecutor(max_workers=2) print('main: starting') f = ex.submit(task, 5) f.arg = 5 f.add_done_callback(done) result = f.result()
不论由于什么原因,只要认为future“完成”,就会调用这个回调,所以在使用它之前必须检查传入回调的对象的状态。
1.5 撤销任务
如果一个future已经提交但还没启动,那么可以调用它的cancel()方法将其撤销。
from concurrent import futures import time def task(n): print('{}: sleeping'.format(n)) time.sleep(0.5) print('{}: done'.format(n)) return n / 10 def done(fn): if fn.cancelled(): print('{}: canceled'.format(fn.arg)) elif fn.done(): print('{}: not canceled'.format(fn.arg)) if __name__ == '__main__': ex = futures.threadpoolexecutor(max_workers=2) print('main: starting') tasks = [] for i in range(10, 0, -1): print('main: submitting {}'.format(i)) f = ex.submit(task, i) f.arg = i f.add_done_callback(done) tasks.append((i, f)) for i, t in reversed(tasks): if not t.cancel(): print('main: did not cancel {}'.format(i)) ex.shutdown()
cancel()返回一个布尔值,指示任务是否可用撤销。
1.6 任务中的异常
如果一个任务产生一个未处理的异常,那么它会被保存到这个任务的future,而且可以通过result()或exception()方法得到。
from concurrent import futures def task(n): print('{}: starting'.format(n)) raise valueerror('the value {} is no good'.format(n)) ex = futures.threadpoolexecutor(max_workers=2) print('main: starting') f = ex.submit(task, 5) error = f.exception() print('main: error: {}'.format(error)) try: result = f.result() except valueerror as e: print('main: saw error "{}" when accessing result'.format(e))
如果在一个任务函数中抛出一个未处理的异常后调用了result(),那么会在当前上下文中再次抛出同样的异常。
1.7 上下文管理器
执行器会与上下文管理器合作,并发的运行任务并等待它们都完成。当上下文管理器退出时,会调用执行器的shutdown()方法。
from concurrent import futures def task(n): print(n) with futures.threadpoolexecutor(max_workers=2) as ex: print('main: starting') ex.submit(task, 1) ex.submit(task, 2) ex.submit(task, 3) ex.submit(task, 4) print('main: done')
离开当前作用域时如果要清理线程或进程资源,那么用这种方式使用执行器就很有用。