Python3标准库：asyncio异步I/O、事件循环和并发工具

1. asyncio异步I/O、事件循环和并发工具

asyncio模块提供了使用协程构建并发应用的工具。threading模块通过应用线程实现并发，multiprocessing使用系统进程实现并发，asyncio则使用一种单线程单进程方法来实现并发，应用的各个部分会彼此合作，在最优的时刻显式地切换任务。大多数情况下，会在程序阻塞等待读写数据时发生这种上下文切换，不过asyncio也支持调度代码在将来的某个特定时间运行，从而支持一个协程等待另一个协程完成，以处理系统信号和识别其他一些事件（这些事件可能导致应用改变其工作内容）。

1.1 异步并发概念

使用其他并发模型的大多数程序都采用线性方式编写，而且依赖于语言运行时系统或操作系统的底层线程或进程管理来适当地改变上下文。基于asyncio的应用要求应用代码显式地处理上下文切换，要正确地使用相关技术，这取决于是否能正确理解一些相关联的概念。asyncio提供的框架以一个事件循环(event loop)为中心，这是一个首类对象，负责高效地处理I/O事件、系统事件和应用上下文切换。目前已经提供了多个循环实现来高效地利用操作系统的功能。尽管通常会自动地选择一个合理的默认实现，但也完全可以在应用中选择某个特定的事件循环实现。在很多情况下这会很有用，例如，在Windows下，一些循环类增加了对外部进程的支持，这可能会以牺牲一些网络I/O效率为代价。与事件循环交互的应用要显式地注册将运行的代码，让事件循环在资源可用时向应用代码发出必要的调用。例如，一个网络服务器打开套接字，然后注册为当这些套接字上出现输入事件时服务器要得到通知。事件循环在建立一个新的进入连接或者在数据可读取时都会提醒服务器代码。当前上下文中没有更多工作可做时，应用代码要再次短时间地交出控制。例如，如果一个套接字再没有更多的数据可以读取，那么服务器会把控制交回给事件循环。
将控制交还给事件循环的机制依赖于Python的协程(coroutine)，这是一些特殊的函数，可以将控制交回给调用者而不丢失其状态。协程与生成器函数非常类似；实际上，在Python3.5版本之前对协程未提供原生支持时，可以用生成器来实现协程。asyncio还为协议(protocol)和传输(transport)提供了一个基于类的抽象层，可以使用回调编写代码而不是直接编写协程。在基于类的模型和协程模型中，可以通过重新进入事件循环显式地改

变上下文，以取代Python多线程实现中隐式的上下文改变。future是一个数据结构，表示还未完成的工作结果。事件循环可以监视Future对象是否完成，从而允许应用的一部分等待另一部分完成一些工作。除了future，asyncio还包括其他并发原语，如锁和信号量。
Task是Future的一个子类，它知道如何包装和管理一个协程的执行。任务所需的资源可用时，事件循环会调度任务运行，并生成一个结果，从而可以由其他协程消费。

1.2 利用协程合作完成多任务

协程是一个专门设计用来实现并发操作的语言构造。调用协程函数时会创建一个协程对象，然后调用者使用协程的send()方法运行这个函数的代码。协程可以使用await关键字（并提供另一个协程)暂停执行。暂停时，这个协程的状态会保留，使得下一次被唤醒时可以从暂停的地方恢复执行。

1.2.1 启动一个协程

asyncio事件循环可以采用多种不同的方法启动一个协程。最简单的方法是使用run_until complete(),并把协程直接传人这个方法。

import asyncioasync def coroutine():print('in coroutine')event_loop = asyncio.get_event_loop()
try:print('starting coroutine')coro = coroutine()print('entering event loop')event_loop.run_until_complete(coro)
finally:print('closing event loop')event_loop.close()

第一步是得到事件循环的一个引用。可以使用默认的循环类型，也可以实例化一个特定的循环类。在这个例子中使用了默认循环。run_until_complete()方法用这个协程启动循环；协程返回退出时这个方法会停止循环。

1.2.2 从协程返回值

协程的返回值传回给启动并等待这个协程的代码。

import asyncioasync def coroutine():print('in coroutine')return 'result'event_loop = asyncio.get_event_loop()
try:return_value = event_loop.run_until_complete(coroutine())print('it returned: {!r}'.format(return_value))
finally:event_loop.close()

在这里，run_unitil_complete()还会返回它等待的协程的结果。

1.2.3 串链协程

一个协程可以启动另一个协程并等待结果，从而可以更容易地将一个任务分解为可重用的部分。下面的例子有两个阶段，它们必须按顺序执行，不过可以与其他操作并发运行。

import asyncioasync def outer():print('in outer')print('waiting for result1')result1 = await phase1()print('waiting for result2')result2 = await phase2(result1)return (result1, result2)async def phase1():print('in phase1')return 'result1'async def phase2(arg):print('in phase2')return 'result2 derived from {}'.format(arg)event_loop = asyncio.get_event_loop()
try:return_value = event_loop.run_until_complete(outer())print('return value: {!r}'.format(return_value))
finally:event_loop.close()

这里使用了await关键字而不是向循环增加新的协程。因为控制流已经在循环管理的一个协程中，所以没有必要告诉循环管理这些新协程。

1.2.4 生成器代替协程

协程函数是asyncio设计中的关键部分。它们提供了一个语言构造，可以停止程序某一部分的执行，保留这个调用的状态，并在以后重新进人这个状态。所有这些动作都是并发框架很重要的功能。
Python3.5引入了一些新的语言特性，可以使用async def以原生方式定义这些协程，以及使用await交出控制，asyncio的例子利用了这些新特性。Python3的早期版本可以使用由 asyncio.coroutine()修饰符包装的生成器函数和yield from来达到同样的效果。

import asyncio@asyncio.coroutine
def outer():print('in outer')print('waiting for result1')result1 = yield from phase1()print('waiting for result2')result2 = yield from phase2(result1)return (result1, result2)@asyncio.coroutine
def phase1():print('in phase1')return 'result1'@asyncio.coroutine
def phase2(arg):print('in phase2')return 'result2 derived from {}'.format(arg)event_loop = asyncio.get_event_loop()
try:return_value = event_loop.run_until_complete(outer())print('return value: {!r}'.format(return_value))
finally:event_loop.close()

前面的例子使用生成器函数而不是原生协程重新实现。

1.3 调度常规函数调用

除了管理协程和/O回调，asyncio事件循环还可以根据循环中保存的一个定时器值来调度常规函数调用。

1.3.1 迅速调度一个回调

如果回调的时间不重要，那么可以用callsoon()调度下一次循环迭代的调用。调用回调时，函数后面额外的位置参数会传入回调。要向回调传入关键字参数，可以使用functools模块的partial(）。

import asyncio
import functoolsdef callback(arg, *, kwarg='default'):print('callback invoked with {} and {}'.format(arg, kwarg))async def main(loop):print('registering callbacks')loop.call_soon(callback, 1)wrapped = functools.partial(callback, kwarg='not default')loop.call_soon(wrapped, 2)await asyncio.sleep(0.1)event_loop = asyncio.get_event_loop()
try:print('entering event loop')event_loop.run_until_complete(main(event_loop))
finally:print('closing event loop')event_loop.close()

回调会按其调度的顺序来调用。

1.3.2 用Delay调度回调

要将一个回调推迟到将来某个时间调用，可以使用call_later()。这个方法的第一个参数是推迟时间(单位为秒)，第二个参数是回调。

import asynciodef callback(n):print('callback {} invoked'.format(n))async def main(loop):print('registering callbacks')loop.call_later(0.2, callback, 1)loop.call_later(0.1, callback, 2)loop.call_soon(callback, 3)await asyncio.sleep(0.4)event_loop = asyncio.get_event_loop()
try:print('entering event loop')event_loop.run_until_complete(main(event_loop))
finally:print('closing event loop')event_loop.close()

在这个例子中，同一个回调函数调度了多次，每次提供了不同的参数。最后一个调用使用了call_soon()，这会在所有按时间调用的实例之前基于参数3来调用这个回调，由此可以看出“迅速”调用的延迟往往最小。

1.3.3 在指定时间内调度一个回调

还可以安排在指定时间内调度一个调用。实现这个目的的循环依赖于一个单调时钟，而不是墙上时钟时间，以确保“now”时间绝对不会逆转。要为一个调度回调选择时间，必须使用循环的time()方法从这个时钟的内部状态开始。

import asyncio
import timedef callback(n, loop):print('callback {} invoked at {}'.format(n, loop.time()))async def main(loop):now = loop.time()print('clock time: {}'.format(time.time()))print('loop  time: {}'.format(now))print('registering callbacks')loop.call_at(now + 0.2, callback, 1, loop)loop.call_at(now + 0.1, callback, 2, loop)loop.call_soon(callback, 3, loop)await asyncio.sleep(1)event_loop = asyncio.get_event_loop()
try:print('entering event loop')event_loop.run_until_complete(main(event_loop))
finally:print('closing event loop')event_loop.close()

需要注意，循环的时间与time.time()返回的值并不一致。

1.4 异步的生成结果

Future表示还未完成的工作的结果。事件循环可以通过监视一个Future对象的状态来指示它已经完成，从而允许应用的一部分等待另一部分完成一些工作。

1.4.1 等待future

Future的做法类似于协程，所以等待协程所用的技术同样可以用于等待future被标记为完成。下面的例子将future传递到事件循环的run_until_complete()方法。

import asynciodef mark_done(future, result):print('setting future result to {!r}'.format(result))future.set_result(result)event_loop = asyncio.get_event_loop()
try:all_done = asyncio.Future()print('scheduling mark_done')event_loop.call_soon(mark_done, all_done, 'the result')print('entering event loop')result = event_loop.run_until_complete(all_done)print('returned result: {!r}'.format(result))
finally:print('closing event loop')event_loop.close()print('future result: {!r}'.format(all_done.result()))

调用set_result()时，Future的状态改为完成，Future实例会保留提供给方法的结果，以备以后获取。

Future还可以结合await关键字使用。

import asynciodef mark_done(future, result):print('setting future result to {!r}'.format(result))future.set_result(result)async def main(loop):all_done = asyncio.Future()print('scheduling mark_done')loop.call_soon(mark_done, all_done, 'the result')result = await all_doneprint('returned result: {!r}'.format(result))event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(event_loop))
finally:event_loop.close()

Future的结果由await返回，所以经常会让同样的代码处理一个常规的协程和一个Future实例。

1.4.2 Future回调

除了做法与协程类似，Future完成时也可以调用回调。回调会按其注册的顺序调用。

import asyncio
import functoolsdef callback(future, n):print('{}: future done: {}'.format(n, future.result()))async def register_callbacks(all_done):print('registering callbacks on future')all_done.add_done_callback(functools.partial(callback, n=1))all_done.add_done_callback(functools.partial(callback, n=2))async def main(all_done):await register_callbacks(all_done)print('setting result of future')all_done.set_result('the result')event_loop = asyncio.get_event_loop()
try:all_done = asyncio.Future()event_loop.run_until_complete(main(all_done))
finally:event_loop.close()

这个回调只希望得到一个参数，即一个Future实例。要想为回调传递额外的参数，可以使用functools.partial()创建一个包装器。

1.5 并发地执行任务

任务是与事件循环交互的主要途径之一。任务可以包装协程，并跟踪协程何时完成。由于任务是Future的子类，所以其他协程可以等待任务，而且每个任务可以有一个结果，在它完成之后可以获取这个结果。

1.5.1 启动一个任务

要启动一个任务，可以使用create_task()创建一个Task实例。只要循环还在运行而且协程没有返回，create_task()得到的任务便会作为事件循环管理的并发操作的一部分运行。

import asyncioasync def task_func():print('in task_func')return 'the result'async def main(loop):print('creating task')task = loop.create_task(task_func())print('waiting for {!r}'.format(task))return_value = await taskprint('task completed {!r}'.format(task))print('return value: {!r}'.format(return_value))event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(event_loop))
finally:event_loop.close()

这个例子中，在main()函数退出之前，会等待任务返回一个结果。

1.5.2 取消一个任务

通过保留create_task()返回的Task对象，可以在任务完成之前取消它的操作。

import asyncioasync def task_func():print('in task_func')return 'the result'async def main(loop):print('creating task')task = loop.create_task(task_func())print('canceling task')task.cancel()print('canceled task {!r}'.format(task))try:await taskexcept asyncio.CancelledError:print('caught error from canceled task')else:print('task result: {!r}'.format(task.result()))event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(event_loop))
finally:event_loop.close()

这个例子会在启动事件循环之前创建一个任务，然后取消这个任务。结果是run_unitl_complete()方法抛出一个CancelledError异常。

如果一个任务正在等待另一个并发运行的操作完成，那么倘若在这个等待时刻取消任务，则其会通过此时产生的一个CancelledError异常通知任务将其取消。

import asyncioasync def task_func():print('in task_func, sleeping')try:await asyncio.sleep(1)except asyncio.CancelledError:print('task_func was canceled')raisereturn 'the result'def task_canceller(t):print('in task_canceller')t.cancel()print('canceled the task')async def main(loop):print('creating task')task = loop.create_task(task_func())loop.call_soon(task_canceller, task)try:await taskexcept asyncio.CancelledError:print('main() also sees task as canceled')event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(event_loop))
finally:event_loop.close()

捕捉异常会提供一个机会，如果必要，可以利用这个机会清理已经完成的工作。

1.5.3 从协程创建任务

ensure_future()函数返回一个与协程执行绑定的Task。这个Task实例再传递到其他代码，这个代码可以等待这个实例，而无须知道原来的协程是如何构造或调用的。

import asyncioasync def wrapped():print('wrapped')return 'result'async def inner(task):print('inner: starting')print('inner: waiting for {!r}'.format(task))result = await taskprint('inner: task returned {!r}'.format(result))async def starter():print('starter: creating task')task = asyncio.ensure_future(wrapped())print('starter: waiting for inner')await inner(task)print('starter: inner returned')event_loop = asyncio.get_event_loop()
try:print('entering event loop')result = event_loop.run_until_complete(starter())
finally:event_loop.close()

需要说明，对于提供给ensure_future()的协程，在使用await之前这个协程不会启动，只有await才会让它执行。

1.6 组合协程和控制结构

一系列协程之间的线性控制流用内置关键字await可以很容易地管理。更复杂的结构可能允许一个协程等待多个其他协程并行完成，可以使用asyncio中的工具创建这些更复杂的结构。

1.6.1 等待多个协程

通常可以把一个操作划分为多个部分，然后分别执行，这会很有用。例如，采用这种方法，可以高效地下载多个远程资源或者查询远程API。有些情况下，执行顺序并不重要，而且可能有任意多个操作，可以使用wait()暂停一个协程，直到其他后台操作完成。

import asyncioasync def phase(i):print('in phase {}'.format(i))await asyncio.sleep(0.1 * i)print('done with phase {}'.format(i))return 'phase {} result'.format(i)async def main(num_phases):print('starting main')phases = [phase(i)for i in range(num_phases)]print('waiting for phases to complete')completed, pending = await asyncio.wait(phases)results = [t.result() for t in completed]print('results: {!r}'.format(results))event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(3))
finally:event_loop.close()

在内部，wait()使用一个set来保存它创建的Task实例，这说明这些实例会按一种不可预知的顺序启动和完成。wait()的返回值是一个元组，包括两个集合，分别包括已完成和未完成的任务。

如果使用wait()时提供了一个超时值，那么达到这个超时时间后，将只保留未完成的操作。

import asyncioasync def phase(i):print('in phase {}'.format(i))try:await asyncio.sleep(0.1 * i)except asyncio.CancelledError:print('phase {} canceled'.format(i))raiseelse:print('done with phase {}'.format(i))return 'phase {} result'.format(i)async def main(num_phases):print('starting main')phases = [phase(i)for i in range(num_phases)]print('waiting 0.1 for phases to complete')completed, pending = await asyncio.wait(phases, timeout=0.1)print('{} completed and {} pending'.format(len(completed), len(pending),))# Cancel remaining tasks so they do not generate errors# as we exit without finishing them.if pending:print('canceling tasks')for t in pending:t.cancel()print('exiting main')event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(3))
finally:event_loop.close()

其余的后台操作要显式地处理，这有多方面的原因。尽管wait()返回时未完成的任务是挂起的，但只要控制返回到事件循环它们就会恢复运行。如果没有另一个wait()调用，则将没有对象接收任务的输出；也就是说，任务会运行并消费资源，但不会带来任何好处。另外，如果程序退出时还有未完成的任务，那么asyncio会发出一个警告。这些警告可能打印到控制台上，应用的用户便会看到。因此，最好取消所有剩余的后台操作，或者使用wait()让它们结束运行。

1.6.2 从协程收集结果

如果后台阶段是明确的，而且这些阶段的结果很重要，那么gather()可能对等待多个操作很有用。

import asyncioasync def phase1():print('in phase1')await asyncio.sleep(2)print('done with phase1')return 'phase1 result'async def phase2():print('in phase2')await asyncio.sleep(1)print('done with phase2')return 'phase2 result'async def main():print('starting main')print('waiting for phases to complete')results = await asyncio.gather(phase1(),phase2(),)print('results: {!r}'.format(results))event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main())
finally:event_loop.close()

gather()创建的任务不会对外提供，所以无法将其取消。返回值是一个结果列表，结果的顺序与传入gather()的参数顺序相同，而不论后台操作实际上是按什么顺序完成的。

1.6.3 后台操作完成时进行处理

as_completed()是一个生成器，会管理指定的一个协程列表，并生成它们的结果，每个协程结束运行时一次生成一个结果。与wait()类似，as_completed()不能保证顺序，不过执行其他动作之前没有必要等待所有后台操作完成。

import asyncioasync def phase(i):print('in phase {}'.format(i))await asyncio.sleep(0.5 - (0.1 * i))print('done with phase {}'.format(i))return 'phase {} result'.format(i)async def main(num_phases):print('starting main')phases = [phase(i)for i in range(num_phases)]print('waiting for phases to complete')results = []for next_to_complete in asyncio.as_completed(phases):answer = await next_to_completeprint('received answer {!r}'.format(answer))results.append(answer)print('results: {!r}'.format(results))return resultsevent_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(3))
finally:event_loop.close()

这个例子启动了多个后台阶段，它们会按其启动顺序的逆序完成。消费生成器时，循环会使用await等待协程的结果。

1.7 同步原语

尽管asyncio应用通常作为单线程的进程运行，不过仍被构建为并发应用。由于I/0以及其他外部事件的延迟和中断，每个协程或任务可能按一种不可预知的顺序执行。为了支持安全的并发执行，asyncio包含了threading和multiprocessing模块中一些底层原语的实现。

1.7.1 锁

Lock可以用来保护对一个共享资源的访问。只有锁的持有者可以使用这个资源。如果有多个请求要得到这个锁，那么其将会阻塞，以保证一次只有一个持有者。

import asyncio
import functoolsdef unlock(lock):print('callback releasing lock')lock.release()async def coro1(lock):print('coro1 waiting for the lock')async with lock:print('coro1 acquired lock')print('coro1 released lock')async def coro2(lock):print('coro2 waiting for the lock')await lock.acquire()try:print('coro2 acquired lock')finally:print('coro2 released lock')lock.release()async def main(loop):# Create and acquire a shared lock.lock = asyncio.Lock()print('acquiring the lock before starting coroutines')await lock.acquire()print('lock acquired: {}'.format(lock.locked()))# Schedule a callback to unlock the lock.loop.call_later(0.1, functools.partial(unlock, lock))# Run the coroutines that want to use the lock.print('waiting for coroutines')await asyncio.wait([coro1(lock), coro2(lock)]),event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(event_loop))
finally:event_loop.close()

锁可以直接调用，使用await来得到，并且使用结束时可以调用release()方法释放锁，如这个例子中的coro2()所示。还可以结合with await关键字使用锁作为异步上下文管理器，如coro1()中所示。

1.7.2 事件

asyncio.Event基于threading.Event。它允许多个消费者等待某个事件发生，而不必寻找一个特定值与通知关联。

import asyncio
import functoolsdef set_event(event):print('setting event in callback')event.set()async def coro1(event):print('coro1 waiting for event')await event.wait()print('coro1 triggered')async def coro2(event):print('coro2 waiting for event')await event.wait()print('coro2 triggered')async def main(loop):# Create a shared eventevent = asyncio.Event()print('event start state: {}'.format(event.is_set()))loop.call_later(0.1, functools.partial(set_event, event))await asyncio.wait([coro1(event), coro2(event)])print('event end state: {}'.format(event.is_set()))event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(event_loop))
finally:event_loop.close()

与Lock一样，coro1()和coro2()会等待设置事件。区别是一旦事件状态改变，它们便可以立即启动，并且它们不需要得到事件对象上的唯一的锁。

1.7.3 条件

Condition的做法与Event类似，只不过不是通知所有等待的协程，被唤醒的等待协程的数目由notify()的一个参数控制。

import asyncioasync def consumer(condition, n):async with condition:print('consumer {} is waiting'.format(n))await condition.wait()print('consumer {} triggered'.format(n))print('ending consumer {}'.format(n))async def manipulate_condition(condition):print('starting manipulate_condition')# pause to let consumers startawait asyncio.sleep(0.1)for i in range(1, 3):async with condition:print('notifying {} consumers'.format(i))condition.notify(n=i)await asyncio.sleep(0.1)async with condition:print('notifying remaining consumers')condition.notify_all()print('ending manipulate_condition')async def main(loop):# Create a conditioncondition = asyncio.Condition()# Set up tasks watching the conditionconsumers = [consumer(condition, i)for i in range(5)]# Schedule a task to manipulate the condition variable
    loop.create_task(manipulate_condition(condition))# Wait for the consumers to be done
    await asyncio.wait(consumers)event_loop = asyncio.get_event_loop()
try:result = event_loop.run_until_complete(main(event_loop))
finally:event_loop.close()

这个例子启动Condition的5个消费者。它们分别使用wait()方法来等待通知让它继续。manipulate_condition()通知一个消费者，再通知两个消费者，然后通知所有其余的消费者。

1.7.4 队列

asyncio.Queue为协程提供了一个先进先出的数据结构，这与线程的queue.Queue或进程的multiprocessing.Queue很类似。

import asyncioasync def consumer(n, q):print('consumer {}: starting'.format(n))while True:print('consumer {}: waiting for item'.format(n))item = await q.get()print('consumer {}: has item {}'.format(n, item))if item is None:# None is the signal to stop.
            q.task_done()breakelse:await asyncio.sleep(0.01 * item)q.task_done()print('consumer {}: ending'.format(n))async def producer(q, num_workers):print('producer: starting')# Add some numbers to the queue to simulate jobsfor i in range(num_workers * 3):await q.put(i)print('producer: added task {} to the queue'.format(i))# Add None entries in the queue# to signal the consumers to exitprint('producer: adding stop signals to the queue')for i in range(num_workers):await q.put(None)print('producer: waiting for queue to empty')await q.join()print('producer: ending')async def main(loop, num_consumers):# Create the queue with a fixed size so the producer# will block until the consumers pull some items out.q = asyncio.Queue(maxsize=num_consumers)# Scheduled the consumer tasks.consumers = [loop.create_task(consumer(i, q))for i in range(num_consumers)]# Schedule the producer task.prod = loop.create_task(producer(q, num_consumers))# Wait for all of the coroutines to finish.await asyncio.wait(consumers + [prod])event_loop = asyncio.get_event_loop()
try:event_loop.run_until_complete(main(event_loop, 2))
finally:event_loop.close()

用put()增加元素和用get()删除元素都是异步操作，因为队列大小可能是固定的（阻塞增加操作），或者队列可能为空（阻塞获取元素的调用）。