Python Async/Await入門指南
轉自：https://zhuanlan.zhihu.com/p/27258289html
本文將會講述Python 3.5以後出現的async/await的使用方法，以及它們的一些使用目的，若是錯誤，歡迎指正。python
昨天看到David Beazley在16年的一個演講：Fear and Awaiting in Async，給了我很多的感悟和啓發，因而想梳理下本身的思路，因此有了如下這篇文章。web
Python在3.5版本中引入了關於協程的語法糖async和await，關於協程的概念能夠先看我在上一篇文章提到的內容。數據庫
看下Python中常見的幾種函數形式：express
1. 普通函數編程
def function(): return 1
2. 生成器函數bash
def generator(): yield 1
在3.5事後，咱們可使用async修飾將普通函數和生成器函數包裝成異步函數和異步生成器。服務器
3. 異步函數（協程）多線程
async def async_function(): return 1
4. 異步生成器app
async def async_generator(): yield 1
經過類型判斷能夠驗證函數的類型
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)
直接調用異步函數不會返回結果，而是返回一個coroutine對象：
print(async_function()) # <coroutine object async_function at 0x102ff67d8>
協程須要經過其餘方式來驅動，所以可使用這個協程對象的send方法給協程發送一個值：
print(async_function().send(None))
不幸的是，若是經過上面的調用會拋出一個異常：
StopIteration: 1
由於生成器/協程在正常返回退出時會拋出一個StopIteration異常，而原來的返回值會存放在StopIteration對象的value屬性中，經過如下捕獲能夠獲取協程真正的返回值：
try: async_function().send(None) except StopIteration as e: print(e.value) # 1
經過上面的方式來新建一個run函數來驅動協程函數：
def run(coroutine): try: coroutine.send(None) except StopIteration as e: return e.value
在協程函數中，能夠經過await語法來掛起自身的協程，並等待另外一個協程完成直到返回結果：
async def async_function(): return 1 async def await_coroutine(): result = await async_function() print(result) run(await_coroutine()) # 1
要注意的是，await語法只能出如今經過async修飾的函數中，不然會報SyntaxError錯誤。
並且await後面的對象須要是一個Awaitable，或者實現了相關的協議。
查看Awaitable抽象類的代碼，代表了只要一個類實現了__await__方法，那麼經過它構造出來的實例就是一個Awaitable：
class 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一個調用異步函數返回的協程對象是合法的。
class 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
接下來是異步生成器，來看一個例子：
假如我要到一家超市去購買土豆，而超市貨架上的土豆數量是有限的：
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)
如今我想要買50個土豆，每次從貨架上拿走一個土豆放到籃子：
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)
對應到代碼中，就是迭代一個生成器的模型，顯然，當貨架上的土豆不夠的時候，這時只可以死等，並且在上面例子中等多長時間都不會有結果（由於一切都是同步的），也許能夠用多進程和多線程解決，而在現實生活中，更應該像是這樣的：
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
當貨架上的土豆沒有了以後，我能夠詢問超市請求須要更多的土豆，這時候須要等待一段時間直到生產者完成生產的過程：
async def ask_for_potato(): await asyncio.sleep(random.random()) all_potatos.extend(Potato.make(random.randint(1, 10)))
當生產者完成和返回以後，這是便能從await掛起的地方繼續往下跑，完成消費的過程。而這整一個過程，就是一個異步生成器迭代的流程：
async def buy_potatos(): bucket = [] async for p in take_potatos(50): bucket.append(p) print(f'Got potato {id(p)}...')
async for語法表示咱們要後面迭代的是一個異步生成器。
def main(): import asyncio loop = asyncio.get_event_loop() res = loop.run_until_complete(buy_potatos()) loop.close()
用asyncio運行這段代碼，結果是這樣的：
Got potato 4338641384... Got potato 4338641160... Got potato 4338614736... Got potato 4338614680... Got potato 4338614568... Got potato 4344861864... Got potato 4344843456... Got potato 4344843400... Got potato 4338641384... Got potato 4338641160... ...
既然是異步的，在請求以後不必定要死等，而是能夠作其餘事情。好比除了土豆，我還想買番茄，這時只須要在事件循環中再添加一個過程：
def main(): import asyncio 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方法。
class 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以後纔有的特性，一樣的還有異步推導表達式，所以在上面的例子中，也能夠寫成這樣：
bucket = [p async for p in take_potatos(50)]
相似的，還有await表達式：
result = [await fun() for fun in funcs if await condition()]
除了函數以外，類實例的普通方法也能用async語法修飾：
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(): t = ThreeTwoOne() await t.begin() print('start')
實例方法的調用一樣是返回一個coroutine：
function = ThreeTwoOne.begin method = function.__get__(ThreeTwoOne, ThreeTwoOne()) import inspect assert inspect.isfunction(function) assert inspect.ismethod(method) assert inspect.iscoroutine(method())
同理還有類方法：
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')
根據PEP 492中，async也能夠應用到上下文管理器中，__aenter__和__aexit__須要返回一個Awaitable：
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)
在3.7版本，contextlib中會新增一個asynccontextmanager裝飾器來包裝一個實現異步協議的上下文管理器：
from contextlib import asynccontextmanager @asynccontextmanager async def get_connection(): conn = await acquire_db_connection() try: yield finally: await release_db_connection(conn)
async修飾符也能用在__call__方法上：
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'))
await和yield from
Python3.3的yield from語法能夠把生成器的操做委託給另外一個生成器，生成器的調用方能夠直接與子生成器進行通訊：
def sub_gen(): yield 1 yield 2 yield 3 def gen(): return (yield from sub_gen()) def main(): for val in gen(): print(val) # 1 # 2 # 3
利用這一特性，使用yield from可以編寫出相似協程效果的函數調用，在3.5以前，asyncio正是使用@asyncio.coroutine和yield from語法來建立協程：
# 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()
然而，用yield from容易在表示協程和生成器中混淆，沒有良好的語義性，因此在Python 3.5推出了更新的async/await表達式來做爲協程的語法。
所以相似如下的調用是等價的：
async with lock: ... with (yield from lock): ... ###################### def main(): return (yield from coro()) def main(): return (await coro())
那麼，怎麼把生成器包裝爲一個協程對象呢？這時候能夠用到types包中的coroutine裝飾器（若是使用asyncio作驅動的話，那麼也可使用asyncio的coroutine裝飾器），@types.coroutine裝飾器會將一個生成器函數包裝爲協程對象：
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()
儘管兩個函數分別使用了新舊語法，但他們都是協程對象，也分別稱做native coroutine以及generator-based coroutine，所以不用擔憂語法問題。
下面觀察一個asyncio中Future的例子：
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()
兩個協程在在事件循環中，協程coro1在執行第一句後掛起自身切到asyncio.sleep，而協程coro2一直等待future的結果，讓出事件循環，計時器結束後coro1執行了第二句設置了future的值，被掛起的coro2恢復執行，打印出future的結果'data'。
future能夠被await證實了future對象是一個Awaitable，進入Future類的源碼能夠看到有一段代碼顯示了future實現了__await__協議：
class 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已經完成：
def 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會執行下面的代碼：
class Future: ... def __iter__(self): ... assert self.done(), "yield from wasn't used with future" return self.result() # May raise too.
最終返回結果給調用方。
前面講了那麼多關於asyncio的例子，那麼除了asyncio，就沒有其餘協程庫了嗎？asyncio做爲python的標準庫，天然受到不少青睞，但它有時候仍是顯得過重量了，尤爲是提供了許多複雜的輪子和協議，不便於使用。
你能夠理解爲，asyncio是使用async/await語法開發的協程庫，而不是有asyncio才能用async/await，除了asyncio以外，curio和trio是更加輕量級的替代物，並且也更容易使用。
curio的做者是David Beazley，下面是使用curio建立tcp server的例子，聽說這是dabeaz理想中的一個異步服務器的樣子：
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))
不管是asyncio仍是curio，或者是其餘異步協程庫，在背後每每都會藉助於IO的事件循環來實現異步，下面用幾十行代碼來展現一個簡陋的基於事件驅動的echo服務器：
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))
驗證一下：
# terminal 1 $ nc localhost 25000 hello world hello world # terminal 2 $ nc localhost 25000 hello world hello world
如今知道，完成異步的代碼不必定要用async/await，使用了async/await的代碼也不必定能作到異步，async/await是協程的語法糖，使協程之間的調用變得更加清晰，使用async修飾的函數調用時會返回一個協程對象，await只能放在async修飾的函數裏面使用，await後面必需要跟着一個協程對象或Awaitable，await的目的是等待協程控制流的返回，而實現暫停並掛起函數的操做是yield。
我的認爲，async/await以及協程是Python將來實現異步編程的趨勢，咱們將會在更多的地方看到他們的身影，例如協程庫的curio和trio，web框架的sanic，數據庫驅動的asyncpg等等...在Python 3主導的今天，做爲開發者，應該及時擁抱和適應新的變化，而基於async/await的協程憑藉良好的可讀性和易用性日漸登上舞臺，看到這裏，你還不趕忙上車？