<jupyter_start><jupyter_text>原文：参考：[PEP 492PEP 525](https://www.python.org/dev/peps/pep-0492/new-standard-library-functions)本文将会讲述Python 3.5之后出现的async/await的使用方法，以及它们的一些使用目的，如果错误，欢迎指正。Python在3.5版本中引入了关于协程的语法糖async和await，关于协程的概念可以先看我在[上一篇](https://zhuanlan.zhihu.com/p/25228075)文章提到的内容。 常见函数形式看下Python中常见的几种函数形式：1. 普通函数<jupyter_code>def function():
    return 1<jupyter_output><empty_output><jupyter_text>2. 生成器函数<jupyter_code>def generator():
     yield 1<jupyter_output><empty_output><jupyter_text>在3.5过后，我们可以使用async修饰将普通函数和生成器函数包装成异步函数和异步生成器。3. 异步函数（协程）<jupyter_code>async def async_function():
    return 1<jupyter_output><empty_output><jupyter_text>4. 异步生成器<jupyter_code>async def async_generator():
    yield 1<jupyter_output><empty_output><jupyter_text>通过类型判断可以验证函数的类型<jupyter_code>import types

print(type(function) is types.FunctionType)
print(type(generator()) is types.GeneratorType)
print(type(async_function()) is types.CoroutineType)
print(type(async_generator()) is types.AsyncGeneratorType)<jupyter_output>True
True
True
True<jupyter_text>直接调用异步函数不会返回结果，而是返回一个coroutine对象：<jupyter_code>print(async_function())<jupyter_output><coroutine object async_function at 0x000002184F42ED48><jupyter_text>协程需要通过其他方式来驱动，因此可以使用这个协程对象的send方法给协程发送一个值：<jupyter_code>print(async_function().send(None))<jupyter_output><empty_output><jupyter_text>不幸的是，如果通过上面的调用会抛出一个异常：`StopIteration: 1`因为生成器/协程在正常返回退出时会抛出一个`StopIteration`异常，而原来的返回值会存放在`StopIteration`对象的`value`属性中，通过以下捕获可以获取协程真正的返回值：<jupyter_code>try:
    async_function().send(None)
except StopIteration as e:
    print(e.value)<jupyter_output>1<jupyter_text>通过上面的方式来新建一个run函数来驱动协程函数：<jupyter_code>def run(coroutine):
    try:
        coroutine.send(None)
    except StopIteration as e:
        return e.value<jupyter_output><empty_output><jupyter_text>异步函数（协程）**在协程函数中，可以通过await语法来主动挂起自身的协程，并等待另一个协程完成直到返回结果：**<jupyter_code>async def async_function():
    return 1

async def await_coroutine():
    result = await async_function()
    print(result)
    
run(await_coroutine())<jupyter_output>1<jupyter_text>要注意的是，**await语法只能出现在通过async修饰的函数中，否则会报`SyntaxError`错误。****而且await后面的对象需要是一个`Awaitable`，或者实现了相关的协议。**查看Awaitable抽象类的代码，表明了**只要一个类实现了`__await__`方法，那么通过它构造出来的实例就是一个`Awaitable`**：```pyclass Awaitable(metaclass=ABCMeta):    __slots__ = ()    @abstractmethod    def __await__(self):        yield    @classmethod    def __subclasshook__(cls, C):        if cls is Awaitable:            return _check_methods(C, "__await__")        return NotImplemented```而且可以看到，`Coroutine`类也继承了`Awaitable`，而且实现了`send`，`throw`和`close`方法。所以`await`一个使用异步的协程对象是合法的。```pyclass Coroutine(Awaitable):    __slots__ = ()    @abstractmethod    def send(self, value):        ...    @abstractmethod    def throw(self, typ, val=None, tb=None):        ...    def close(self):        ...            @classmethod    def __subclasshook__(cls, C):        if cls is Coroutine:            return _check_methods(C, '__await__', 'send', 'throw', 'close')        return NotImplemented```  异步生成器 接下来是异步生成器，来看一个例子：假如我要到一家超市去购买土豆，而超市货架上的土豆数量是有限的：<jupyter_code>class Potato:
    @classmethod
    def make(cls, num, *args, **kws):
        potatos = []
        for i in range(num):
            potatos.append(cls.__new__(cls, *args, **kws))
        return potatos

all_potatos = Potato.make(5)<jupyter_output><empty_output><jupyter_text>现在我想要买50个土豆，每次从货架上拿走一个土豆放到篮子：<jupyter_code># 示例代码，不要调用
def take_potatos(num):
    count = 0
    while True:
        if len(all_potatos) == 0:
            sleep(.1)
        else:
            potato = all_potatos.pop()
            yield potato
            count += 1
            if count == num:
                break

def buy_potatos():
    bucket = []
    for p in take_potatos(50):
        bucket.append(p)<jupyter_output><empty_output><jupyter_text>对应到代码中，就是迭代一个生成器的模型，显然，当货架上的土豆不够的时候，这时只能够**死等**，而且在上面例子中等多长时间都不会有结果（因为一切都是同步的），也许可以用多进程和多线程解决，而在现实生活中，更应该像是这样的：<jupyter_code>async def take_potatos(num):
    count = 0
    while True:
        if len(all_potatos) == 0:
            await ask_for_potato()
        potato = all_potatos.pop()
        yield potato
        count += 1
        if count == num:
            break<jupyter_output><empty_output><jupyter_text>当货架上的土豆没有了之后，我可以询问超市请求需要更多的土豆，这时候需要等待一段时间直到生产者完成生产的过程：<jupyter_code>async def ask_for_potato():
    await asyncio.sleep(random.random())
    all_potatos.extend(Potato.make(random.randint(1, 10)))<jupyter_output><empty_output><jupyter_text>当生产者完成和返回之后，这是便能从await挂起的地方继续往下跑，完成消费的过程。而这整一个过程，就是一个异步生成器迭代的流程：<jupyter_code>async def buy_potatos():
    bucket = []
    async for p in take_potatos(50):
        bucket.append(p)
        print(f'Got potato {id(p)}...')<jupyter_output><empty_output><jupyter_text>`async for`语法表示我们要后面迭代的是一个异步生成器。```pydef main():    import asyncio    loop = asyncio.get_event_loop()    res = loop.run_until_complete(buy_potatos())    loop.close()```结果是这样的```Got potato 2283605158712...Got potato 2283605158656...Got potato 2283605157704...Got potato 2283604970072...Got potato 2283354273720...Got potato 2283606583616...Got potato 2283606583560......``` 完整示例代码```pyimport asyncioimport randomclass Potato:    @classmethod    def make(cls, num, *args, **kws):        potatoes = []        for i in range(num):            potatoes.append(cls.__new__(cls, *args, **kws))        return potatoesall_potatoes = Potato.make(5)async def take_potatoes(num):    count = 0    while True:        if len(all_potatoes) == 0:            await ask_for_potato()        yield all_potatoes.pop()        count += 1        if count == num:            breakasync def ask_for_potato():    await asyncio.sleep(random.random())    all_potatoes.extend(Potato.make(random.randint(1, 10)))async def buy_potatoes():    bucket = []    async for p in take_potatoes(50):        bucket.append(p)        print(f'Got potato {id(p)}...')def main():    loop = asyncio.get_event_loop()    loop.run_until_complete(buy_potatoes())    loop.close()if __name__ == '__main__':    main()``` 多个异步任务既然是异步的，在请求之后不一定要死等，而是可以做其他事情。比如除了土豆，我还想买番茄，这时只需要在事件循环中再添加一个过程：```pyimport asynciodef main():    loop = asyncio.get_event_loop()    res = loop.run_until_complete(asyncio.wait([buy_potatos(), buy_tomatos()]))    loop.close()```再来运行这段代码：```Got potato 4423119312...Got tomato 4423119368...Got potato 4429291024...Got potato 4421640768...Got tomato 4429331704...Got tomato 4429331760...Got tomato 4423119368...Got potato 4429331760...Got potato 4429331704...Got potato 4429346688...Got potato 4429346072...Got tomato 4429347360......```  异步生成器的定义看下`AsyncGenerator`的定义，它需要实现`__aiter__`和`__anext__`两个核心方法，以及`asend`，`athrow`，`aclose`方法。```pyclass AsyncGenerator(AsyncIterator):    __slots__ = ()    async def __anext__(self):        ...    @abstractmethod    async def asend(self, value):        ...    @abstractmethod    async def athrow(self, typ, val=None, tb=None):        ...    async def aclose(self):        ...    @classmethod    def __subclasshook__(cls, C):        if cls is AsyncGenerator:            return _check_methods(C, '__aiter__', '__anext__',                                  'asend', 'athrow', 'aclose')        return NotImplemented```          常见用法 异步推导式异步生成器是在3.6之后才有的特性，同样的还有异步推导表达式，因此在上面的例子中，也可以写成这样：```pybucket = [p async for p in take_potatos(50)]```类似的，还有await表达式：```pyresult = [await fun() for fun in funcs if await condition()]```  装饰类方法除了函数之外，类实例的普通方法也能用async语法修饰：<jupyter_code>import asyncio


class ThreeTwoOne:
    async def begin(self):
        print(3)
        await asyncio.sleep(1)
        print(2)
        await asyncio.sleep(1)
        print(1)        
        await asyncio.sleep(1)
        return

async def game():
    await ThreeTwoOne.begin()
    print('start')
    

# loop = asyncio.get_event_loop()
# loop.run_until_complete(game())
# loop.close()
ThreeTwoOne.begin
function.__get__(ThreeTwoOne, ThreeTwoOne())
function = ThreeTwoOne.begin
method = function.__get__(ThreeTwoOne, ThreeTwoOne())

import inspect
assert inspect.isfunction(function)
assert inspect.ismethod(method)
assert inspect.iscoroutine(method())<jupyter_output>D:\SoftInstall\Anacoda3\lib\site-packages\ipykernel_launcher.py:7: RuntimeWarning: coroutine 'ThreeTwoOne.begin' was never awaited
  import sys
RuntimeWarning: Enable tracemalloc to get the object allocation traceback<jupyter_text>同理还有类方法：<jupyter_code>class ThreeTwoOne:
    @classmethod
    async def begin(cls):
        print(3)
        await asyncio.sleep(1)
        print(2)
        await asyncio.sleep(1)
        print(1)
        await asyncio.sleep(1)
        return


async def game():
    await ThreeTwoOne.begin()
    print('start')<jupyter_output><empty_output><jupyter_text>上下文管理器根据PEP 492中，`async`也可以应用到上下文管理器中，`__aenter__`和`__aexit__`需要返回一个`Awaitable`：<jupyter_code>import asyncio


class GameContext:
    async def __aenter__(self):
        print('game loading...')
        await asyncio.sleep(1)

    async def __aexit__(self, exc_type, exc, tb):
        print('game exit...')
        await asyncio.sleep(1)


async def game():
    async with GameContext():
        print('game start...')
        await asyncio.sleep(2)


# loop = asyncio.get_event_loop()
# loop.run_until_complete(game())
# loop.close()<jupyter_output><empty_output><jupyter_text>在3.7版本，contextlib中会新增一个asynccontextmanager装饰器来包装一个实现异步协议的上下文管理器：<jupyter_code>from contextlib import asynccontextmanager


@asynccontextmanager
async def get_connection():
    conn = await acquire_db_connection()
    try:
        yield
    finally:
        await release_db_connection(conn)<jupyter_output><empty_output><jupyter_text>`async`修饰符也能用在`__call__`方法上：<jupyter_code>import asyncio
from time import time


class GameContext:

    async def __aenter__(self):
        self._started = time()
        print('game loading...')
        await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print('game exit...')
        await asyncio.sleep(1)

    async def __call__(self, *args, **kws):
        if args[0] == 'time':
            return time() - self._started


async def game():
    async with GameContext() as ctx:
        print('game start...')
        await asyncio.sleep(2)
        print('game time: ', await ctx('time'))

# loop = asyncio.get_event_loop()
# loop.run_until_complete(game())
# loop.close()<jupyter_output><empty_output><jupyter_text>await和yield from`await`和`yield from`Python3.3的`yield from`语法可以把生成器的操作委托给另一个生成器，生成器的调用方可以直接与子生成器进行通信：<jupyter_code>def sub_gen():
    yield 1
    yield 2
    yield 3

def gen():
    return (yield from sub_gen())

def main():
    for val in gen():
        print(val)

main()<jupyter_output>1
2
3<jupyter_text>利用这一特性，使用`yield from`能够编写出类似协程效果的函数调用，在3.5之前，`asyncio`正是使用`@asyncio.coroutine`和`yield from`语法来创建协程<jupyter_code># https://docs.python.org/3.4/library/asyncio-task.html
import asyncio

@asyncio.coroutine
def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    yield from asyncio.sleep(1.0)
    return x + y

@asyncio.coroutine
def print_sum(x, y):
    result = yield from 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()<jupyter_output><empty_output><jupyter_text>**然而，用`yield from`容易在表示协程和生成器中混淆，没有良好的语义性**，所以在Python 3.5推出了更新的async/await表达式来作为协程的语法。因此类似以下的调用是等价的：```pyasync with lock:    ...    with (yield from lock):    ...def main():    return (yield from coro())def main():    return (await coro())```    那么，怎么把生成器包装为一个协程对象呢？这时候可以用到`types`包中的`coroutine`装饰器（如果使用`asyncio`做驱动的话，那么也可以使用`asyncio.coroutine`装饰器），`@types.coroutine`装饰器会将一个生成器函数包装为协程对象：<jupyter_code>import asyncio
import types

@types.coroutine
def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    yield from 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()<jupyter_output><empty_output><jupyter_text>尽管两个函数分别使用了新旧语法，但他们都是协程对象，也分别称作`native coroutine`以及`generator-based coroutine`，因此不用担心语法问题。下面观察一个asyncio中Future的例子：<jupyter_code>import asyncio

future = asyncio.Future()

async def coro1():
    await asyncio.sleep(1)
    future.set_result('data')

async def coro2():
    print(await future)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait([
    coro1(), 
    coro2()
]))
loop.close()<jupyter_output><empty_output><jupyter_text>两个协程在在事件循环中，协程coro1在执行第一句后挂起自身切到`asyncio.sleep`，而协程coro2一直等待future的结果，让出事件循环，计时器结束后coro1执行了第二句设置了future的值，被挂起的coro2恢复执行，打印出future的结果'data'。future可以被`await`证明了future对象是一个`Awaitable`，进入Future类的源码可以看到有一段代码显示了future实现了`__await__`协议：```pyclass Future:    ...    def __iter__(self):        if not self.done():            self._asyncio_future_blocking = True            yield self   This tells Task to wait for completion.        assert self.done(), "yield from wasn't used with future"        return self.result()   May raise too.    if compat.PY35:        __await__ = __iter__  make compatible with 'await' expression```当执行await future这行代码时，future中的这段代码就会被执行，首先future检查它自身是否已经完成，如果没有完成，挂起自身，告知当前的Task（任务）等待future完成。当future执行set_result方法时，会触发以下的代码，设置结果，标记future已经完成：```pydef set_result(self, result):    ...    if self._state != _PENDING:        raise InvalidStateError('{}: {!r}'.format(self._state, self))    self._result = result    self._state = _FINISHED    self._schedule_callbacks()```    最后future会调度自身的回调函数，触发Task._step()告知Task驱动future从之前挂起的点恢复执行，不难看出，future会执行下面的代码：```pyclass Future:    ...    def __iter__(self):        ...        assert self.done(), "yield from wasn't used with future"        return self.result()   May raise too.```最终返回结果给调用方。  其他协程库：curio前面讲了那么多关于`asyncio`的例子，那么除了`asyncio`，就没有其他协程库了吗？`asyncio`作为python的标准库，自然受到很多青睐，但它有时候还是显得太重量了，尤其是提供了许多复杂的轮子和协议，不便于使用。你可以理解为，`asyncio`是使用`async/await`语法开发的协程库，而不是有`asyncio`才能用`async/await`，除了`asyncio`之外，`curio`和`trio`是更加轻量级的替代物，而且也更容易使用。`curio`的作者是<NAME>，下面是使用`curio`创建Tcp server的例子，据说这是dabeaz理想中的一个异步服务器的样子：<jupyter_code>from curio import run, spawn
from curio.socket import *


async def echo_server(address):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind(address)
    sock.listen(5)
    print('Server listening at', address)
    async with sock:
        while True:
            client, addr = await sock.accept()
            await spawn(echo_client, client, addr)


async def echo_client(client, addr):
    print('Connection from', addr)
    async with client:
        while True:
            data = await client.recv(100000)
            if not data:
                break
            await client.sendall(data)
    print('Connection closed')


if __name__ == '__main__':
    run(echo_server, ('', 25000))<jupyter_output><empty_output><jupyter_text>验证一下：```$ nc localhost 25000hello worldhello world``` 无论是`asyncio`还是`curio`，或者是其他异步协程库，在背后往往都会借助于IO的事件循环来实现异步，下面用几十行代码来展示一个简陋的基于事件驱动的echo服务器：<jupyter_code>from socket import socket, AF_INET, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR
from selectors import DefaultSelector, EVENT_READ

selector = DefaultSelector()
pool = {}


def request(client_socket, addr):
    client_socket, addr = client_socket, addr

    def handle_request(key, mask):
        data = client_socket.recv(100000)
        if not data:
            client_socket.close()
            selector.unregister(client_socket)
            del pool[addr]
        else:
            client_socket.sendall(data)

    return handle_request


def recv_client(key, mask):
    sock = key.fileobj
    client_socket, addr = sock.accept()
    req = request(client_socket, addr)
    pool[addr] = req
    selector.register(client_socket, EVENT_READ, req)


def echo_server(address):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind(address)
    sock.listen(5)
    selector.register(sock, EVENT_READ, recv_client)
    try:
        while True:
            events = selector.select()
            for key, mask in events:
                callback = key.data
                callback(key, mask)
    except KeyboardInterrupt:
        sock.close()


if __name__ == '__main__':
    echo_server(('', 25000))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12427638871782239, 'fraction_numerical': 0.02346348072422712, 'mean_word_length': 4.228267868641339, 'pattern_counts': {'":': 0, '<': 101, '<?xml version=': 0, '>': 101, 'https://': 3, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '26', 'dir': 'jupyter-structured-clean-dedup', 'id': '351895', 'license': 'MIT', 'n_tokens_mistral': 7011, 'n_tokens_neox': 7097, 'n_words': 952, 'path': '并发&异步/协程（四）- Async&Await入门指南.ipynb', 'repo_name': 'wanZzz6/Modules-Learn', 'size': '16245'}