万一免五开户,联系微信:golangsysu

python asyncio 异步的精髓

日常分享 数学王子秦老师 383℃

协程框架基础

任何一个协程框架都首先必须是一个异步框架,asyncio也不例外。一个异步框架通常主要包括事件循环、事件队列、polling、timer队列,所有的异步框架皆不例外,asyncio也是如此。事件循环是实际启动之后执行的代码,事件队列用来向事件循环发送要执行的任务,polling使用multiplexing技术(如select或epoll)用来监控socket等IO活动,timer队列保存定时器,一般是个最小堆。

执行过程以asyncio为例,asyncio的事件队列里面就是普通的callable,callable执行的时候调用asyncio的其他接口配置polling或者timer队列或者发送更多callable到事件队列里。polling或者timer队列会将socket活动和计时器关联到不同的回调,这个回调不是立即执行的,而是延迟回调,也就是不直接调用而是放进事件队列里,等着事件循环去调用。事件循环(也就是所谓EventLoop)开始的时候,会不断从事件队列里取callable,然后一个一个call过去,call完换下一个;事件队列里没有了,就去timer队列里取一个最近的timer作为超时时间,去调用polling,直到任意socket活动,或者超时为止,然后根据不同的条件,将socket的回调放到事件队列里,从头开始执行。这个过程描述下来,是不是感觉很简单?

那这么简单的结构怎么实现异步编程呢?原理也很简单,每个callable都是一小份工作,当这部分工作做完之后,会等待下一个条件:如果要等socket,就设置polling回调;如果要等超时时间,就设置timer;如果要等其他callable完成,就使用同步对象的callback接口(后面会讲的Future);如果要同时等多个条件,就把所有的回调都设置上。通过这些回调不断触发,就能实现异步编程的目的。

基于回调的代码结构在这件事情上就比较直白,直接将回调函数设置成polling或者timer或者同步对象的callback就行了。

Future基础

Future是什么呢?基本的Future就是一个回调管理器,任何程序都可以通过add_callback的接口为这个Future设置一个回调函数,在Future完成时会调用这个回调函数。Future刚创建的时候是未完成状态,有一个接口set_result可以将Future设置为完成状态,很显然实际上也是这个set_result接口调用了已经注册的那些callback,很简单很弱智(simple and stupid,stupid对于设计来说是一个褒义词)的设计,说穿了就是宣布Future完成的那个callable自己去调了这些回调。如果add_callback的时候Future已经完成,就让add_callback自己去直接调用了这个回调,这样效果就统一了。这其中,唯一的秘密在于,Future调用callback并不是直接调用(否则反复触发会导致栈溢出),而是通过前面说的延迟调用,将callable放到事件队列里,让事件循环帮忙调用一下。

使用Future的好处在于可以方便地管理回调——也就只有这么点用了。顺便它可以保存一下结果,回调的时候能读出来这个Future之前是成功了还是失败了。

Future可以cancel,cancel实际上是为Future设置了一个特殊的“已取消”的结果,其他跟普通的结果没有多大区别。

Future可以串联或并联起来,串联的Future使用一个callback自动触发后一个Future,这样这两个Future就会依次完成,可以用来将不同的流程连接起来;并联起来的多个Future,可以在全部完成,或者第一个完成之类的情况下触发另一个Future,实现一些较复杂的流程控制。

Coroutine与Task

下一个问题是回调结构,说白了,我们不太喜欢[callback hell](https://www.zhihu.com/search?q=callback hell&search_source=Entity&hybrid_search_source=Entity&hybrid_search_extra={"sourceType"%3A"answer"%2C"sourceId"%3A"555273313"})——一段复杂的异步代码需要写非常多的callback闭包,管理状态的时候很麻烦。怎么解决呢?Python有一种结构叫做生成器,也就是带有yield的函数,叫做生成器函数,它在调用的时候会返回一个生成器(generator),这个生成器可以步进——也就是说,调用一下next,就走一小段,暂停下来,返回一个值;再调用一下,又走一小段,返回一个值。这个结构就很有意思了,我们可以用它来代替普通的callback,每次callback的时候步进一小段就好了,它的结构让它自己就可以保存之前的过程的状态,在外部看来是异步(每次调用一个callable),在它自己看来却是一个连续的同步过程,这样就可以用类似于同步的写法来写异步过程了。

更妙的是,Python中的生成器不仅可以每次返回一个值,它还可以再接收一个值,也可以在暂停的位置直接抛出异常,这就有很丰富的流程控制体验了。

我们将生成器跟Future结合起来,就可以设计一种叫做coroutine的异步流程,它是一个生成器,每次通过yield返回一个Future,然后我们为这个Future设置一个callback,这个callback触发的时候,我们将Future里保存的结果发送给这个coroutine,这会让coroutine往下再步进一步,得到下一个Future,这样依次执行下去,我们就可以让这个coroutine在我们的异步模型当中以一种类似于同步的方式执行下去。

要做到这个目的,我们需要有一个壳套在生成器外面,它用来自动调用Future的add_callback,做一些coroutine和事件循环之间的适配的工作,这个壳就叫做asyncio.Task,把一个coroutine放到一个Task的壳里面,coroutine就可以在事件循环里相对独立地执行了。asyncio.Task自己也是一个Future的子类,这样也可以把asyncio.Task作为一个Future来使用,就可以等待一个任务执行结束了;区别在于,Task的结果是自动用coroutine返回值设置的,而且Task的cancel不仅会设置“已取消”的结果,还会立即在coroutine中抛出异常,尝试使coroutine提前结束。

除了使用Task作为Future来串联,coroutine之间还可以级联——在一个coroutine内,想要执行另一个coroutine,只需要将这个coroutine的yield值和返回值,以及外部发送的值全部都代理到内层coroutine里就行了。在Python 3中有一个专门的[yield from](https://www.zhihu.com/search?q=yield from&search_source=Entity&hybrid_search_source=Entity&hybrid_search_extra={"sourceType"%3A"answer"%2C"sourceId"%3A"555273313"})的语法用来做这件事。

从Python 3.5开始,coroutine有了一个专门的async def的写法,将coroutine和普通的生成器区分开。async def中的await语义和yield有些区别,它更接近于yield from,因此如果await一个coroutine,实际上是前面说的级联;如果await一个Future,Future通过await方法在内部yield了自己,这样就跟直接yield一个Future没太大区别了,由此所有的逻辑都可以通过简单的await语法来实现。

处理polling和timer时也很简单,方法实际上是创建一个新的Future,用polling或者timer的回调来设置Future的结果,剩下的就可以回到使用Future的方法上。

举例

async def do_something():
    await asyncio.sleep(2)
    r = await some_other_thing()
    return r + 1

这是一个典型的coroutine函数,调用这个函数得到一个coroutine对象。这个coroutine首先等待了一个Future(asyncio.sleep返回一个定时器相关的Future),然后级联了另一个coroutine方法,最后返回了一个结果。使用loop.run_until_complete()就可以执行这个异步过程了,也可以在其他异步过程中await这个过程,或者使用 asyncio.ensure_future()异步执行这个过程。

超级精髓的总结

  1. EventLoop负责执行事件循环整个过程,调用各个回调函数
  2. Future是回调管理器
  3. Coroutine使用生成器技术来替代连续的多个回调
  4. Task负责将Coroutine接口和Future、EventLoop接口对接起来,同时它自己也是一个Future
  5. 别想太多,async def + await走起就对了,asyncio很简单,凭直觉写出来的大部分都是对的

转载请注明:吾要开户 » python asyncio 异步的精髓

喜欢 (0)or分享 (0)