为了更好地理解 Python 的异步编程(asynchronous programming),本文将通过 PEP(Python增强建议书)的路线图(从 Python 2.2 中 PEP 255 引入生成器开始,直到最新的 Python 3.6 中 PEP 525/530 对异步生成器/推导式的支持),来详述梳理 Python 异步模型(asynchronous paradigm)的前世今生,其中主要涉及:生成器、协程、asyncio 库、async/await 等概念和语法。

测试环境:macOS 10.13.3;Python 3.6.5。

PEP 255:简易生成器

  • 在 Python 2.2 中,首次引入了「PEP 255 -- Simple Generators」中的生成器(generators)的概念。由于它也实现了迭代器协议(iterator protocol),因此也称为生成器迭代器(generator iterators),其设计初衷就是为了在不浪费内存的情况下迭代计算下一个值。

容器(container)是一种包含元素的数据结构,且它会将所包含的所有元素都存放在内存中。Python 中常见的容器有:列表 list、集合 set、字典 dict、元组 tuple、字符串 str 等。我们之所以能访问容器中的每一个元素,是因为绝大部分容器都是可迭代的(iterable)。此外,文件句柄、sockets 等也都是可迭代的。

  • 可迭代对象指的是能够返回迭代器(iterator)的对象。下例中,x 是一个可迭代对象(列表),通过 iter() 方法,便可以将一个可迭代对象转换为迭代器。
x = [1, 2, 3]
type(x)  # <class 'list'>
y = iter(x)
type(y)  # <class 'list_iterator'>
  • 迭代器(iterator)是一个带状态的对象,通过调用 next() 方法便可以得到下一个值。一个实现了 __iter__() 方法的类是可迭代的,该方法会返回对象本身(即迭代器);而一个实现了 __next__() 方法的类则是迭代器。通常,一个可迭代类会同时实现这两种方法。
next(y)  # 1
next(y)  # 2

迭代器协议指的是对象必须提供一个 next() 方法,返回迭代过程中的下一个值,或者引起一个 StopIteration 异常,以终止迭代。

  • 生成器(generator)是一种特殊的迭代器。它不需要实现 __iter__()__next__() 方法,而是通过关键字 yield 来表征。

常见的两种生成器是:1)生成器函数,即含有关键字 yield 的函数;2)生成器表达式,它类似于列表推导式(list comprehension),不同之处在于生成器表达式返回的是一个生成器,而不是常规容器的推导式。

numbers = [1, 2, 3, 4, 5, 6]
l = [x * x for x in numbers]
type(l)  # <class 'list'>
d = {x: x * x for x in numbers}
type(d)  # <class 'dict'>
g = (x * x for x in numbers)
type(g)  # <class 'generator'>

注意:生成器表达式返回的不是元组推导式。

Ys9NMaubp44hghEi.png

PEP 342:基于改进生成器的协程

  • 在 Python 2.5 中,「PEP 342 -- Coroutines via Enhanced Generators」为生成器引入了一个极其重要的功能:不仅能够通过 yield 暂停生成器,还能在暂停时通过 send() 方法给生成器传递数据。不同于生成器函数中的 yield 语句,此时 yield 关键字被用作为表达式(即出现在 = 右边)。
def coro():
    text = yield "Hello"
    yield text

c = coro()
print(next(c))          # Hello
print(c.send("World"))  # World
  • 支持通过 send() 方法回传数据到生成器的函数称为基于生成器的协程(generator based coroutines),可以用于实现简易的协程。这也是 Python 中协程的雏形。到目前为止,「生成器」和「协程」并没有本质上的区别。但自 Python 3.5 引入了 asyncawait 关键字后,Python 便开始了支持原生的协程。相关内容将在后文详述。

「Python Cookbook」的作者 David Beazley 在「A Curious Course on Coroutines and Concurrency」中指出,基于生成器的协程是 Python 书籍中最少被提及的 Python 特性。

PEP 380:子生成器的委托

理解 yield from 机制的最好方法是把它看作是调用者和子生成器中的一个透明的双向通道,参见 Praveen Gollakota 的「回答」。

  • 当在一个生成器中调用另一个(子)生成器时,需要通过 for x in subgenerator: yield x 来「重新」yield 子生成器的值。而 yield from subgenerator 可以简化这种操作,从「」方向来说,由此可以形成链式生成器。
def subgenerator():
    for j in range(10):
        yield j

# Delegator using *yield*
def generator_v1():
    for i in subgenerator():
        yield i

# Delegator using *yield from*
def generetor_v2():
    yield from subgenerator()

注意:不能简单地把 yield from 看做是用来简化原始 for 循环的一个语法糖(syntactic sugar)。事实上,yield from 包含了对 PEP 342 中提到的很多机制的处理,如 .throw().send().close() 等。

  • 从「」方向来说,Praveen Gollakota 给出了一个用 yield from 封装子生成器的实现。封装器 writer_wrapper,即中间层的委托生成器,既可以接收来自调用者(caller)wrap 的输入(通过 send() 方法),同时也可以向子生成器发送数据。
def writer():
    while True:
        w = (yield)
        print('>> ', w)

def writer_wrapper(coro):
    yield from coro

w = writer()
wrap = writer_wrapper(w)
wrap.send(None)  # "prime" the coroutine
for i in range(4):
    wrap.send(i)

小结:Python 2.2 中的生成器(generators)使得代码的执行能够被暂停;Python 2.5 允许向挂起的生成器发送数据,这使得协程(coroutines)成为了可能;Python 3.3 的 yield from 使我们能更容易地重构生成器,并形成链式生成器。


咖啡时间:事件循环

  • 理解事件循环(event loop)有助于更好地理解异步编程(asynchronous programming)和 Python 3.4 中的 asyncawait。简单来说,事件循环就是一个等待和分派事件或消息的一个系统。

之所以称之为事件循环,是因为它会一直收集新的事件,并遍历寻找有没有对应的事件处理函数。

  • 以浏览器中的 Javascript 事件循环为例:当我们点击(click)一个按钮时,便会进入 Javascript 的事件循环。事件循环会检查是否注册过能够响应该事件的回调函数(callback)。如果存在(这里通常是 onclick() 函数),便会执行对应的回调函数。

而 Python 则是通过 asyncio 标准库来支持事件循环的。


PEP 492:使用 asyncawait 语法的协程

  • 在 Python 3.4 中,使用模块 asyncio 便可以进行通用的异步编程(async programming),尽管它使用的还是基于生成器的协程(使用 yield from)。

这里,异步编程指的是编写的代码其执行顺序事先未知;而并发编程(concurrent programming)指的是编写能够独立执行的代码,即使是在同一个线程之中。

import asyncio

@asyncio.coroutine
def countdown(number, n):
    while n > 0:
        print('T-minus', n, '({})'.format(number))
        yield from asyncio.sleep(1)
        n -= 1

loop = asyncio.get_event_loop()
tasks = [
    asyncio.ensure_future(countdown("A", 2)),
    asyncio.ensure_future(countdown("B", 3))]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

装饰器 @asyncio.coroutine 用来标记一个使用 asyncio 和事件循环的协程(这是 Python 第一次对协程有明确的定义)。asyncio.ensure_future() 方法用于调度协程至事件循环中,之后便可以启动该事件循环。

  • 上例中,事件循环会执行每个协程,直到遇到 yield from。它告知事件循环,这是一个需要等待的操作,并返回一个 asyncio.Future 对象给事件循环,此时事件循环便会挂起协程的执行。然后,事件循环负责监测每一个 future 对象(这里便是 asyncio.sleep(1) 函数),一旦 future 对象执行完成,事件循环便会把结果重新发回协程中,使其能够继续执行,最后直到所有的协程都执行完毕,事件循环也就没有了需要监测的 future 对象了。

上例代码引自「Curio - A Tutorial Introduction」,其中使用了 asyncio.wait() 方法。从功能上来说,asyncio.gather()asyncio.wait() 十分相似,但前者是在较高层级上对任务的组合,而后者则提供更多底层的操作,如参数 return_when=asyncio.FIRST_COMPLETED 可以指明当第一个任务完成后便停止等待。具体的差别可以参考 Udi 在「Asyncio.gather vs asyncio.wait」中的回答。

XQRdqCuNzhydeNcc.png

  • 在 Python 3.5 中,「PEP 492 -- Coroutines with async and await syntax」引入了新的 asyncawait 语法来支持原生协程(native coroutine),也称为基于 async 的协程,即:使用 async def 来定义一个协程,在协程内部则使用 await 来代替 yield from

官方给出了一个很好的「链式协程的例子」,以及对应的时序图。

import asyncio

async def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(1.0)
    return x + y

async def print_sum(x, y):
    result = await compute(x, y)
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()

PFHDjQk8EZfM7ml0.png

注意:基于生成器的协程和原生协程两者不能混用,即:不能在原生协程定义中使用 yieldyield from 关键字,也不能在基于生成器的协程中使用 await 关键字。

  • 尽管这里 awaityield from 语法上没什么区别,但 await 除了能修饰协程以外,还能修饰一类称为「可异步等待的」对象(awaitable object),它实现了 __await__() 方法,并返回一个迭代器。

进一步,可以从汇编层面讨论两者的差别,具体按可参见「How the heck does async/await work in Python 3.5?」一文。作者同时指出,只有基于生成器的协程才能真正暂停执行,并强制性返回给事件循环,而原生协程则不可以。因为在最底层的协程中,肯定是使用 yield 或者 yield from,而如果只工作在上层的话,使用 await 即可。

PEP 525:异步生成器

注意:异步生成器不是协程,它没有实现 __await__() 方法,因此也不能被 await。PEP 525 中指出,异步生成器要比等价实现的异步迭代器的性能高出两倍

async def ticker(delay, to):
    """Yield numbers from 0 to *to* every *delay* seconds."""
    for i in range(to):
        yield i
        await asyncio.sleep(delay)
result = [i async for i in aiter() if i % 2]

注意:异步推导式只能在使用 async def 定义的函数中使用,且 aiter() 必须是一个实现了 __aiter__() 方法的对象,或者另一个 async def 定义的函数,如下例所示。

import asyncio
 
async def numbers(numbers):
    for i in range(numbers):
        yield i
        await asyncio.sleep(0.5)
 
async def main():
    odd_numbers = [i async for i in numbers(10) if i % 2]
    print(odd_numbers)
 
if __name__ == '__main__':
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(main())
    event_loop.close()
  • 至此,便基本理解了生成器、协程、基于生成器的协程、原生协程、异步生成器等概念。最后,给出一个实际使用了异步生成器的代码实例。
import asyncio
import ccxt.async as ccxt

async def poll(tickers):
    i = 0
    kraken = ccxt.kraken()
    while True:
        symbol = tickers[i % len(tickers)]
        yield (symbol, await kraken.fetch_ticker(symbol))
        i += 1
        await asyncio.sleep(kraken.rateLimit / 1000)

async def main():
    async for (symbol, ticker) in poll(['BTC/USD', 'ETH/BTC', 'BTC/EUR']):
        print(symbol, ticker)

asyncio.get_event_loop().run_until_complete(main())

发表新评论

沪ICP备17018959号-3