理解python异步机制

2016-09-13 本文已影响2287人 treelake

1. python yield与async/await

要点

最重要的是生成器函数碰到yield停止执行，收到next或send才会继续执行的机制。
而且send方法令我们可以传递值到生成器暂停的地方。
生成器执行结束抛出StopIteration 异常。
yield from用于把其他生成器当做子例程调用。

MUST: yield from + iterable
MUST: @asyncio.coroutines + yield from / yield
MUST: await + coroutine / object with await() method (that must ruturn a iterable instead of a coroutine)
MUSTN'T: async def + yield/yield from

图说

以下两张图中是6组代码，实现了同样的控制程序执行顺序的功能，调度器一样，输出也完全一样，几个小的演变一定程度上展示了yield和async/await的关系。

由yield到async/await

由yield到async/await变化标示

下面这张图是用python内部的审查函数来判断类型，查看上图中的@coroutine+yield构成的函数到底在python内部被当做什么

@coroutine+yield到底是什么

2. async/await 实验

这是一系列关于await，async特性的小实验，该新特性在PEP-0492中描述并在Python3.5中实现。是一个相当不错的便于初学者理解await，async的资源。
在这里我翻译了作者的纲要并阐述了自己的理解，不当之处欢迎指正。
代码参看github/awaitexp。

实验一：最简调度器

目的是有一个基本上最简单可实现的协程调度器。它与标准库中asyncio模块中相对复杂的调度器并列。
首先创建一个最简单的基于生成器的协程函数A

@types.coroutine
def switch():
    yield

然后它可以被其他用async def定义的的协程函数B和C await，只有当await返回时，B和C才继续执行。
这样我们就可以有效地控制B和C的执行顺序。
然后我们创建了一个调度器，它对列表进行了两次深拷贝以避免问题。它循环协程队列，使用send方法对每个协程依次递进，如果有协程已经完成则将其移出队列，当列表中的协程全部完成时结束。

实验二：具有睡眠功能的简单调度器

目的是给我们的调度器添加一个awaitable的睡眠协程函数。尽管睡眠功能本身是相当简单的，这还是给调度器增加了一定的复杂度。之后该调度器的睡眠功能是否能简易地与其他事件组合将是一个有趣的问题。
我们在第一个实验的基础上首先修改switch函数使其yield返回输入给它的参数的字典，并且添加一个新的调用它的协程函数。

async def sleep(delay):
    await switch(delay=delay)

然后通过args=coro.send(None)与该函数碰撞，得到含有delay参数的字典作为send的返回值。便可以判断出是否调用调度器的睡眠机制。
最后在调度器中实现每一次协程列表循环结束后判断在睡眠列表中的协程是否有到时间的，到时间或时间超出则添加到运行协程列表中进入循环执行。如果运行列表中的协程都执行完了，则查看睡眠列表中的协程中还需睡眠的最少时间，线程睡眠，睡眠完成再将其添加到运行队列。

实验三：使能增加新的协程

当前只有在开始指定的协程能被调度。该实验的目的是允许新的协程能够在调度器开始运行后被加入。
在当前实验中我们拓展调度器为一个类并持有自己的协程队列，并提供一个添加协程进入队列的方法。然后获取调度器的唯一实例，在定义的协程中使用实例方法来加入新的协程，再把该协程加入调度，这样就实现了在运行中的某一刻加入新的协程。

实验四：和线程池进行交互

有时我们会有一些计算（或者I/O）任务，它们最好能在背景中执行，而不是在主线程中。这个实验展示了一个示例，通过线程池执行器和被调度的协程交互来获得背景计算。从长远看，这个技术很可能不是一个良好的基础，因为它不能同时伺服IO选择器和线程队列。
在这一歩实验中我们主要是添加了两个部分，第一部分是一个装饰器:

def background(fn):  
    @wraps(fn)
    async def wrapper(*args,**kwargs):
        return await switch(op='background',fn=fn,args=args,kwargs=kwargs)
    return wrapper

该装饰器能将一个比较耗时的计算函数封装为一个协程，使其可以被其他协程await。在调度器中利用send函数的返回值可以获取它的类型为background、函数入口地址以及函数的传参，然后在调度器中按相应机制执行。
第二部分是在调度器中的修改：我们让调度器类拥有了一个私有的concurrent.futures.ThreadPoolExecutor()对象。并在运行协程队列的循环判断中将background类型的操作提交给线程池对象，并将当前的协程移出运行队列，添加到futures队列中。然后在每次运行队列循环后判断futures中的任务是否有完成的（使用的参数为一旦有任一任务完成或被取消都返回），如果主线程此时处于将要睡眠的状态，就等待相应的时间，没有的话则立刻返回，下次再查询，完成的任务将其所在协程带入运行队列，任务结果通过调度器send传回该协程。

实验五：

协程的一个典型应用就是和异步I/O交互。该实验的目的是让调度器和Python的selector模型进行交互并提供一个基本的I/O模型。
类似于实验四，我们在实验三的基础上让调度器拥有了一个私有的selectors.DefaultSelector()对象。创建了一个名为io辅助协程，供其他协程调用并且给调度器提供类型和操作信息。同样利用主线程睡眠时间来等待selector的select方法。关键语句：for key, events in self.selector.select(timeout=timeout): ...，select返回在超时时间内IO准备好的对象列表。
此外一部分是创建服务端和客户端的协程。
在服务端的server协程函数中首先初始化socket，然后在主循环中await io操作（服务器本身是一个协程！），利用io辅助协程来等待io资源，取得连接后调用echo函数处理该连接，并将其加入调度器。在echo协程中，同样利用io辅助协程异步获取读写权限。循环直到break结束关闭链接。
对于客户端测试程序，需要另外获取一个调度器。定义一个echoclient协程函数，利用io辅助协程异步发送接收信息。最后将所有测试协程加入调度器列表，开始执行。

部分结果，可以看出异步执行的效果

实验六：

目的是有效地合并实验四和实验五的特性。这将使用标准的self-pipe（实际上是一个self-socket）技巧来通知背景中复杂计算方法的完成。
在服务端的echo协程函数中添加了await add(3, 5)语句，add函数中添加sleep代表耗时的背景计算。在调度器类中添加了自己的一对socket。当有复杂计算时，添加入futures队列，同时设置future完成回调函数，该回调函数向自己的socket发送提示完成的消息。在select方法时判断是否有准备好的io描述符属于自己拥有的socket，如果有就意味着一个计算的完成，然后清空socket缓存，移除相应的futures。

程序结构示意图