并发-异步-协程-asyncio in Python, 2022

(2022.07.06 Tues)
注：本文翻译自realpython网站的async-io-python，找了多份相关文章，这是我个人感觉最清晰的对异步的解释文。翻译并保留在简书。

提到协程，首先需要对同步、异步、多线程、多进程、并发、并行等概念有大概的了解。

概念

并发(concurrency)与并行(parallelism)的比较：

并行代表了多个操作在同时间运行，多进程(multiprocessing)是实现并行的手段，而且将多个任务分摊给CPU。多进程适用于CPU密集型任务(CPU-bound tasks)，for循环和数学计算常属于此类。

并发是比并行更宽泛的概念，并发的多任务以混叠方式(overlapping manner)运行。多线程(threading)是并发的执行模型，多个线程轮流执行任务。一个进程可包含多个线程。Python与线程的关系参考GIL，本文不讨论。

多线程适合于IO密集型任务(IO-bound tasks)。CPU密集型任务以CPU持续工作为特征，而IO密集型任务中有大量的等待时间，等待输入/输出的完成。

综上(to recap the above)，并发包含了多进程(适用于CPU密集型任务)和多线程(IO密集型任务)。多进程是并行的一种形式，而并行又是并发的特殊形式/子类。Python标准库对这些技术都提供了支持，multiprocessing，threading和concurrent.futures等包。

下面引入异步IO(asynchronous IO, async IO)概念。异步IO不是新概念，已经在其他语言中如C#、Go、Scala中有所发展。Python中的异步IO使用asyncio包和async、await关键字。

异步IO不是多线程或多进程，并且并不基于这两者。异步IO基于单进程、单线程设计，它使用协同多任务(cooperative multitasking)。也就是说异步IO给人一种并发的错觉但使用的仍然是单一进程中的单一线程。协程(coroutines)，作为异步IO的核心特征，可以并发执行(scheduled concurrently)，但其本质并非并发。

需要强调，异步IO是并发编程的一种方式，但异步IO不是并行。它更像多线程而非多进程，但它和多线程多进程都不同，是并发工具包中的独立一员。

至于何为异步(asynchronous)，这里给出两个属性：

• 异步流程(routine)可以暂停并在等待最终结果时让其他流程运行
• 异步编码方便了并发执行，换句话说，异步呈现并发的状态

对比多线程和协程
线程是进程的一个实体，是CPU调度的基本单位，是比进程Process更小的能独立运行的基本单位。每个线程不拥有资源，除了运行中必不可少的计数器、寄存器和栈等，每个线程与同一个进程中的其他线程共享进程所拥有的的全部资源。线程间通信主要通过共享内存。其上下文切换快，资源开销少，不稳定易丢失数据。

协程Coroutine，是轻量级线程，其调度由用户控制。协程拥有自己的寄存器，上下文和栈。协程调度切换时，将寄存器、上下文和栈保存到其他地方。在切换回来之时，恢复之前保存的寄存器、上下文和栈；而操作栈则没有内核切换开销，可不加锁访问全局变量，因此上下文切换速度很快。

总结：
并发概念包含并行概念，并行是并发的特例。多进程是实现并行的方法，多进程和异步IO是并发的实现方法。

Async IO

考虑象棋一对n的车轮大战。主棋手和每个对手下棋，下棋过程中如果主旗手在和每个棋手对弈时都在自己出手后等待对方再出手，然后来到下一个对手前，则这是同步方式。如果主旗手出手后不等待对方出手而直接进入下一个对手的棋盘上，则是异步方式。同步方式完成和所有人的对弈时间将远大于异步方式的时间。

因此协同多任务是可以当做是程序事件循环，也就是通过与多任务通信使得每个任务轮流运行。

异步IO等待时间更长，在等待期间，函数被阻塞，并允许其他函数运行。A function that blocks effectively forbids others from running from the time that it starts until the time that it returns.

创建高效稳定的多线程代码比较难且容易出错，而异步IO避免了多线程设计中可能遇到的屏障。Python的异步模型建立在如下这些概念之上，比如回调(callback)，事件(events)，传播(transports)，协议和future。持续变化的API使得这些概念并不容易理解。幸运的是Python中的asyncio这个专门用于异步IO的包已经走向成熟，其中的大多数特性不再改变(provisional)。

asyncio包和async、await关键字

异步IO的核心是协程(coroutines)。协程是一个特殊版本的Python生成器函数，该函数可以在到达return语句之前挂起(suspend)其本身的执行，并可以间接的将控制权临时交给其他协程。

下面是异步IO的hello world代码，让我们感受异步IO。

import time
import uuid
import random
import asyncio
import logging
import threading

async def count():
    t = threading.currentThread()
    i = uuid.uuid1()
    logging.info(f"""ONE. Thread id: {t.ident}, uuid: {i}""")
    await asyncio.sleep(1)
    logging.info(f"""TWO. Thread id: {t.ident}, uuid: {i}""")

async def main():
    await asyncio.gather(count(), count(), count())

if __name__ == "__main__":
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO) 
    s = time.perf_counter()
    asyncio.run(main())
    elapsed = time.perf_counter() - s
    print(f""" executed in {elapsed:0.2f} seconds.""")

运行结果如下

2022-10-29 12:11:09,403: ONE. Thread id: 4613266880, uuid: b7a6419e-573f-11ed-8473-acde48001122
2022-10-29 12:11:09,403: ONE. Thread id: 4613266880, uuid: b7a64be4-573f-11ed-8473-acde48001122
2022-10-29 12:11:09,403: ONE. Thread id: 4613266880, uuid: b7a64e64-573f-11ed-8473-acde48001122
2022-10-29 12:11:10,408: TWO. Thread id: 4613266880, uuid: b7a6419e-573f-11ed-8473-acde48001122
2022-10-29 12:11:10,408: TWO. Thread id: 4613266880, uuid: b7a64be4-573f-11ed-8473-acde48001122
2022-10-29 12:11:10,408: TWO. Thread id: 4613266880, uuid: b7a64e64-573f-11ed-8473-acde48001122
 executed in 1.01 seconds.

数据结果的顺序是异步IO的心脏。每次调用count()都是一个单独的事件循环(event loop)，或协调者(coordinator)。当每个任务执行到await asyncio.sleep(1)语句，函数将控制权给事件循环，像是在说"我睡一秒钟，你快用这事件忙你的事吧"。

作为对比，同步方式的代码如下

import time

def count():
    print("One")
    time.sleep(1)
    print("Two")

def main():
    for _ in range(3):
        count()

if __name__ == "__main__":
    s = time.perf_counter()
    main()
    elapsed = time.perf_counter() - s
    print(f"executed in {elapsed:0.2f} seconds.")

执行结果为

One
Two
One
Two
One
Two
executed in 3.01 seconds.

使用time.sleep()和asyncio.sleep()看起来并无区别，他们用来提到耗时任务(time-intensive)的过程。time.sleep()代表了任何耗时的阻塞函数的调用，而asyncio.sleep()用于替代非阻塞调用，尽管同样需要花费时间完成。

后面你将看到，asyncio.sleep()命令和await命令的好处在于调用该命令的函数可以将控制权割让(cede)给其他马上可以执行的函数。对比之下，time.sleep()以及其他阻塞调用(blocking call)不兼容python异步代码，因为该指令阻碍了其完成之前其他函数的运行。

(2022.10.30 Sun)
另一个案例，查看不同协程运行时等待不同的时间，返回结果的顺序

import time 
import uuid
import logging
import asyncio

async def count(l):
    i = str(uuid.uuid1()).split('-')[0]
    logging.info(f"""befor, uuid: {i}, to sleep {l} sec""")
    await asyncio.sleep(l)
    logging.info(f"""after, uuid: {i}, wake up from {l} sec""")

async def main():
    await asyncio.gather(count(2), count(1), count(3))

if __name__ == "__main__":
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO) 
    s = time.perf_counter()
    asyncio.run(main())
    elapsed = time.perf_counter() - s
    print(f""" executed in {elapsed:0.2f} seconds.""")

返回结果

$ python app_async.py 
2022-10-30 12:07:51,782: befor, uuid: 6c456b5c, to sleep 2 sec
2022-10-30 12:07:51,782: befor, uuid: 6c45787c, to sleep 1 sec
2022-10-30 12:07:51,783: befor, uuid: 6c458024, to sleep 3 sec
2022-10-30 12:07:52,787: after, uuid: 6c45787c, wake up from 1 sec
2022-10-30 12:07:53,786: after, uuid: 6c456b5c, wake up from 2 sec
2022-10-30 12:07:54,785: after, uuid: 6c458024, wake up from 3 sec
executed in 3.00 seconds.

注意到不同的协程在事件池中循环时，一旦await的协程没有产生结果，则不会阻塞。推测继续在事件池中循环查看哪个协程有结果，并按先后顺序展示结果。

异步IO规则

语法async def可以引入原生协程(native coroutine)，也可以引入异步生成器(asynchronous generator)。async with和async for也同样有效。

await关键字将函数控制权返还给事件循环，同时悬挂(suspend)当前函数/协程的执行。Python代码中遇到函数g()的中表达式await f()，这代表着await指令告诉事件循环：暂停函数g()的执行，直到f()返回结果，而我等的就是f()的结果，与此同时让其他程序运行吧。典型结构如下

async def g():
    # Pause here and come back to g() when f() is ready
    r = await f()
    return r

下面是一些使用async、await的场景

• 使用了关键字async def标识的函数是一个协程(coroutine)。它还可以使用await、return、yield关键字，但不是必须。仅使用async def noop(): pass也可以
  • 使用await和/或return关键字创建协程函数，为了调用该协程函数，必须使用await等其结果。
  • 在async def的代码块(code block)中使用yield关键字创建异步生成器，尽管这不太常见。该异步生成器可以用async for来迭代。这个技术暂不讨论(forget about async generators for the time being)。
  • 使用了async def关键字的部分无法使用yield from关键字，否则返回SyntaxError错误。
• await关键字用在async def协程之外的地方会产生SyntaxError错误，如果yield用在def函数之外也会产生同样的错误。

用下面代码解释上面提到的规则

async def f(x):
    y = await z(x)  # OK - `await` and `return` allowed in coroutines
    return y

async def g(x):
    yield x  # OK - this is an async generator

async def m(x):
    yield from gen(x)  # No - SyntaxError

def m(x):
    y = await z(x)  # Still no - SyntaxError (no `async def` here)
    return y

当你使用await f()命令，要求f()是可被等待的对象(awaitable object)，也就是1) 另一个协程，或2) 定义了.__await__()方法并返回迭代器的对象。多数情况下只需要关心第1种情况。

说到这里，不得不提到一种将函数标记为协程的古老方法：用@asyncio.coroutine装饰一个def函数。装饰后返回一个基于装饰器的协程(generator-based coroutine)。这种用法在Python3.5出现async/await之后被定为过时。下面这两种协程的定义方法等价，差别在于第一种是基于生成器的协程，第二种是原生协程(native coroutine)。

import asyncio

@asyncio.coroutine
def py34_coro():
    """Generator-based coroutine, older syntax"""
    yield from stuff()

async def py35_coro():
    """Native coroutine, modern syntax"""
    await stuff()

基于生成器的协程有其特定的规则，将在Python 3.10被移除，开发时请优先使用原生协程。async/await方法使得协程成为Python的一个独立特性，可减少概念模糊。

下面这个案例展示了异步IO如何节省等待时间：给定一个协程makerandom()，该协程持续生成[0, 10]之间的随机整数，直到任意一个超出预定门限，对该协程进行多次调用，彼此之间不需等待对方完成。

import asyncio
import random

# ANSI colors
c = (
    "\033[0m",   # End of color
    "\033[36m",  # Cyan
    "\033[91m",  # Red
    "\033[35m",  # Magenta
)

async def makerandom(idx: int, threshold: int = 6) -> int:
    print(c[idx + 1] + f"Initiated makerandom({idx}).")
    i = random.randint(0, 10)
    while i <= threshold:
        print(c[idx + 1] + f"makerandom({idx}) == {i} too low; retrying.")
        await asyncio.sleep(idx + 1)
        i = random.randint(0, 10)
    print(c[idx + 1] + f"---> Finished: makerandom({idx}) == {i}" + c[0])
    return i

async def main():
    res = await asyncio.gather(*(makerandom(i, 10 - i - 1) for i in range(3)))
    return res

if __name__ == "__main__":
    random.seed(444)
    r1, r2, r3 = asyncio.run(main())
    print()
    print(f"r1: {r1}, r2: {r2}, r3: {r3}")

返回结果如图

coroutine demo results
该程序使用了主协程makerandom()，以三个不同的输入并行运行该协程。多数程序都会包含小的模块化的协程和一个包裹函数(wrapper function)，该包裹函数用于将小的协程连在一起。main()方法用将任务聚集在一起(futures)，通过在池或迭代上映射中心协程。

该案例中，池是range(3)。在后文的完整案例中，池是一系列即将被并行请求、解析和处理的URLs，而main()封装了每个URL的例行程序(routine)。

生成随机数是一个CPU密集型任务，不是asyncio的最好的案例，而asyncio.sleep()在案例中被设计用来模拟IO密集型任务。

(2022.07.07 Thur)

异步IO设计模式

异步IO的设计基本上可以分为链式协程和Queue协程。

链式协程 chaining coroutine

协程的关键特性之一在于协程可以链式连接在一起。一个协程对象是可等待的(awaitable)，所以另一个协程可以等待这个协程对象。这个特性可以保证将程序分为更小、更易管理、可回收的协程。

import asyncio
import random
import time

async def part1(n: int) -> str:
    i = random.randint(0, 10)
    print(f"part1({n}) sleeping for {i} seconds.")
    await asyncio.sleep(i)
    result = f"result{n}-1"
    print(f"Returning part1({n}) == {result}.")
    return result

async def part2(n: int, arg: str) -> str:
    i = random.randint(0, 10)
    print(f"part2{n, arg} sleeping for {i} seconds.")
    await asyncio.sleep(i)
    result = f"result{n}-2 derived from {arg}"
    print(f"Returning part2{n, arg} == {result}.")
    return result

async def chain(n: int) -> None:
    start = time.perf_counter()
    p1 = await part1(n)
    p2 = await part2(n, p1)
    end = time.perf_counter() - start
    print(f"-->Chained result{n} => {p2} (took {end:0.2f} seconds).")

async def main(*args):
    await asyncio.gather(*(chain(n) for n in args))

if __name__ == "__main__":
    import sys
    random.seed(444)
    args = [1, 2, 3] if len(sys.argv) == 1 else map(int, sys.argv[1:])
    start = time.perf_counter()
    asyncio.run(main(*args))
    end = time.perf_counter() - start
    print(f"Program finished in {end:0.2f} seconds.")

运行结果如下

part1(1) sleeping for 4 seconds.
part1(2) sleeping for 4 seconds.
part1(3) sleeping for 0 seconds.
Returning part1(3) == result3-1.
part2(3, 'result3-1') sleeping for 4 seconds.
Returning part1(1) == result1-1.
part2(1, 'result1-1') sleeping for 7 seconds.
Returning part1(2) == result2-1.
part2(2, 'result2-1') sleeping for 4 seconds.
Returning part2(3, 'result3-1') == result3-2 derived from result3-1.
-->Chained result3 => result3-2 derived from result3-1 (took 4.00 seconds).
Returning part2(2, 'result2-1') == result2-2 derived from result2-1.
-->Chained result2 => result2-2 derived from result2-1 (took 8.00 seconds).
Returning part2(1, 'result1-1') == result1-2 derived from result1-1.
-->Chained result1 => result1-2 derived from result1-1 (took 11.01 seconds).
Program finished in 11.01 seconds.

在这个案例中，main()的运行时间等于任务中最长的运行时间。

队列式协程 Queue coroutine

Python的asyncio包提供了queue class，该queue被设计成与Python的queue相似的类。到目前为止我们还没有用过队列结构。前面的链式协程中，每个任务(future)有一些列协程组成，每个协程等待彼此并向对方传递单一输入。

异步IO的另一种结构是，若干互不关联的生产者(producers)向同一个队列中加入item。每个生产者随机地、无通知地向队列中加入元素。一组消费者贪婪地从队列中拉取元素，不需等待任何信号。

在这个模式中，没有消费者到生产者的链式连接。消费者不知道生产者数量，也不知道队列中有多少对象。队列在这里相当于一个吞吐量中介，它联通了生产者和消费者而不需要双方直接沟通。

在多线程编程中，线程安全一直是需要关注的问题，因此线程安全的queue.Queue()经常被采用，但是在异步IO中这个问题不需要担心，除非将异步IO和多线程编程结合编程。本案例中，队列扮演的是生产者和消费者的天线，使得他们不需要直接链式连接或关联。

下面案例的同步版本看起来效率低下：若干阻塞性生产者连续地向queue中添加元素，一次一个生产者。只有当所有生产者都完成后消费者才能开始处理queue，每次只有一个消费者，逐一处理。这种设计的延迟严重，且queue中的元素只能静静等待而无法被尽快的选择与处理。

异步版本如下面代码所示。工作流(workflow)中最有挑战的地方在于消费者需要一个信号，用以告知生产者已经完成了(将元素插入queue)。不然await q.get()将会无限期悬挂(hang)，因为queue已经被完全处理，但消费者无从得知生产者已经完成工作。

main()中，关键点在于await q.join()该指令阻塞直到queue中的所有元素都被处理完成，之后取消消费者任务，否则hang up并无限期等待queue中出现新的元素。

import asyncio
import itertools as it
import os
import random
import time

async def makeitem(size: int = 5) -> str:
    return os.urandom(size).hex()

async def randsleep(caller=None) -> None:
    i = random.randint(0, 10)
    if caller:
        print(f"{caller} sleeping for {i} seconds.")
    await asyncio.sleep(i)

async def produce(name: int, q: asyncio.Queue) -> None:
    n = random.randint(0, 10)
    for _ in it.repeat(None, n):  # Synchronous loop for each single producer
        await randsleep(caller=f"Producer {name}")
        i = await makeitem()
        t = time.perf_counter()
        await q.put((i, t))
        print(f"Producer {name} added <{i}> to queue.")

async def consume(name: int, q: asyncio.Queue) -> None:
    while True:
        await randsleep(caller=f"Consumer {name}")
        i, t = await q.get()
        now = time.perf_counter()
        print(f"Consumer {name} got element <{i}>"
              f" in {now-t:0.5f} seconds.")
        q.task_done()

async def main(nprod: int, ncon: int):
    q = asyncio.Queue()
    producers = [asyncio.create_task(produce(n, q)) for n in range(nprod)]
    consumers = [asyncio.create_task(consume(n, q)) for n in range(ncon)]
    await asyncio.gather(*producers)
    await q.join()  # Implicitly awaits consumers, too
    for c in consumers:
        c.cancel()

if __name__ == "__main__":
    import argparse
    random.seed(444)
    parser = argparse.ArgumentParser()
    parser.add_argument("-p", "--nprod", type=int, default=5)
    parser.add_argument("-c", "--ncon", type=int, default=10)
    ns = parser.parse_args()
    start = time.perf_counter()
    asyncio.run(main(**ns.__dict__))
    elapsed = time.perf_counter() - start
    print(f"Program completed in {elapsed:0.5f} seconds.")

运行

_StoreAction(option_strings=['-p', '--nprod'], dest='nprod', nargs=None, const=None, default=5, type=<class 'int'>, choices=None, help=None, metavar=None)
_StoreAction(option_strings=['-c', '--ncon'], dest='ncon', nargs=None, const=None, default=10, type=<class 'int'>, choices=None, help=None, metavar=None)
Producer 0 sleeping for 4 seconds.
Producer 2 sleeping for 7 seconds.
Consumer 0 sleeping for 7 seconds.
Consumer 1 sleeping for 8 seconds.
...
Producer 0 added <4c215641e9> to queue.
Producer 0 sleeping for 10 seconds.
Producer 3 added <7ad6097316> to queue.
Producer 3 sleeping for 0 seconds.
...
Consumer 1 sleeping for 10 seconds.
Program completed in 40.01253 seconds.

前几个协程是帮手函数(helper functions)，返回随机字符，小数形式的性能计数器和随机整数。生产者将1到5之间的随机整数放入队列queue。每个插入的元素是一个tuple，其中的i是一个随机字符，t是生产者将该tuple放入queue的时间戳。

当消费者从queue中拉取一个元素，消费者将计算该元素在queue中的存留时间。

asyncio.sleep()命令用于其他更加复杂且耗时的协程，并且这些协程如果是常规阻塞函数，将阻塞其他函数执行。

这个案例中，延迟的产生有可能有两个原因：

• 标准的，基本不可避免的overhead
• 当queue中有元素时所有消费者都在休眠

关于第二个原因，当消费者规模达到数百数千时会相当正常。在不同的系统中，会有不同的用户控制生产者和消费者的管理，而queue作为中央吞吐(central throughput)而存在。

截止到这里，你已经了解了异步IO调用协程的基本和若干案例。

异步IO与生成器的关系：以生成器为原型

前面我们对比过基于生成器的协程和原生协程，比如下面这个案例

import asyncio

@asyncio.coroutine
def py34_coro():
    """Generator-based coroutine"""
    # No need to build these yourself, but be aware of what they are
    s = yield from stuff()
    return s

async def py35_coro():
    """Native coroutine, modern syntax"""
    s = await stuff()
    return s

async def stuff():
    return 0x10, 0x20, 0x30

如果仅仅调用各自的函数名，而不适用await或asyncio.run()等命令，我们得到如下结果

>> py35_coro()
<coroutine object py35_coro at 0x7fecbe9f1140>
>> py34_coro()
<stdin>:1: RuntimeWarning: coroutine 'py35_coro' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
<generator object py34_coro at 0x7fecbea08c10>

注意到生成器调用与协程调用的结果相似，都是没有直接结果，而返回对象。事实上协程的底层是增强版的生成器(enhanced generator)。技术上，await指令更像是yield from指令的近似(而非yield)，不过也要keep in mind，yield from x()是for i in x(): yield i命令的语法糖而已。

生成器与异步IO的一个重要特性是可以根据自身需要暂停和重启。比如可以break生成器对象中的迭代，并以余下的值重启迭代。当生成器函数遇到yield，该函数'产生'yield命令的值，之后进入空闲状态(idle)直到下一次再被要求'产生'后续的值。比如下面这个例子

>>> from itertools import cycle
>>> def endless():
...     """Yields 9, 8, 7, 6, 9, 8, 7, 6, ... forever"""
...     yield from cycle((9, 8, 7, 6))

>>> e = endless()
>>> total = 0
>>> for i in e:
...     if total < 30:
...         print(i, end=" ")
...         total += i
...     else:
...         print()
...         # Pause execution. We can resume later.
...         break
9 8 7 6 9 8 7 6 9 8 7 6 9 8

>>> # Resume
>>> next(e), next(e), next(e)
(6, 9, 8)

而await关键字在行为上表现非常相似，该关键字标记了一个暂停点(break point)，协程运行到暂停点则暂停(suspend)并让其他协程工作。这里的暂停，指的是协程暂时让出控制权，但并没有完全退出和结束。yield、yield from和await都用来在生成器的执行过程中标记一个break point。

上面提到的特性也是生成器和函数的核心差别。函数总是执行所有或不执行(all-or-nothing)，一旦开始执行将不会停止，直到遇到return命令，并返回一个返回值给调用者(caller)。而生成器，在每次遇到yield命令时暂停，不仅要将该命令相关的值推送到调用栈(calling stack)，同时保留局部变量用于在调用next()时返回结果。

生成器还有一个不太广为人知(lesser-known)的特性同样重要。通过.send()命令可以向生成器中传递值，它允许生成器(或协程)调用(或await)其他对象而不需要阻塞。Python的PEP 342中有更多技术细节。

总结协程和生成器的特点：

• 协程是重新利用的生成器，并充分利用了生成器的特性
• 基于生成器的协程使用yield from等待协程结果。使用原生协程的现代Python语法用await代替yield from命令实现相同的功能，即等待协程记过。
• await也可作为中断点使用(break point)，它可暂停协程执行并允许稍后返回。

Other Features: async for and Async Generators + Comprehensions

skip

事件循环和asyncio.run()

事件循环可以被认为是while True循环，用以监控协程，看有无可以执行的东西等等。它可以唤醒一个空闲协程。到目前为止，事件循环的管理都被一个函数调用控制：

asyncio.run(main())  # Python 3.7+

asyncio.run()在Python 3.7中引入，用于负责获取事件循环，运行任务直到所有任务标记为完成，并关闭循环。

另有一个啰嗦的(long-winded)方式管理asyncio事件循环，即用get_event_loop()命令，典型用法如下

loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

尽管在以前的案例中会出现loop.get_event_loop()的调用方式，除非你对事件循环的管理有特别的控制需求，asyncio.run()命令对多数程序来说已经足够。

如果你需要在Python程序中与事件循环做交互，loop是个不错的Python对象，它支持loop.is_running()和loop.is_closed()。该对象可以用于精确控制，比如通过将loop设定为参数安排一个callback。

关于事件循环，有如下几点需要关注：

• 协程与事件循环联合使用时才会更有价值，单独使用协程用处不太大
  主协程等待其他协程，单独调用的话基本没效果：

>>> async def main():
...     print('1')
...     await asyncio.sleep(1)
...     print('2')
... 
>>> main()
<coroutine object main at 0x7fecbe9f1140>

只有使用了asyncio.run()命令才会真正驱动main()协程的执行，以时间循环的方式：

>>> asyncio.run(main())
1
2

其他协程可以靠await执行。典型应用是将main()包裹进asyncio.run()，且使用await指令的链式协程可以在这里调用。

• 默认情况下，异步IO事件循环在单一CPU的单一线程中运行，一般来说这已经足够了。当然也可以在多核中运行事件循环，代价是电脑资源将会占满。

• 事件循环可以自定义(pluggable)，也就是说你甚至可以自己实现事件循环并运行任务，与协程的结构无关。asyncio包包含了两种事件循环的实现，默认的是基于selectors包，另一种实现转为Windows订制。

一个完整的程序：异步请求 Asynchronous Requests

这里我们查看一个有趣、无痛的案例，建立一个URL网络爬虫器，使用aiohttp，建立一个超快的异步HTTP客户端/服务器端架构，当然我们只需要实现客户端。这个工具可用于映射一簇网站的链接，而链接形成一个有向图。

Python的requests包不兼容异步IO，requests的包建立在urllib3之上，urllib3又基于Python的http和socket包。默认情况下，socket操作是阻塞的。这意味着Python不会出现await request.get(url)指令因为.get()方法不是可等待的。相比之下，aiohttp是个可等待协程，比如session.request()和response.text()。requests是个很伟大的包，但将其应用于异步编码你是自己找虐(you are doing yourself a disservice by using requests in asynchronous code)。

程序架构如下：

• 从本地文件，i.e., urls.txt，中读取一些列URLs
• 向URLs发送GET请求并解码返回内容，如果这步失败，则暂停这个URL的请求
• 从响应的HTML的href tags中查找URLs
• 将结果写入文件foundurls.txt

尽可能以异步和并发的方式完成上面的所有步骤，提示使用aiohttp作为requests，用aiofiles做文件叠加(file-appends)。这两个主要的IO案例适合异步IO模型。

首先来看urls.txt文件中保存了哪些链接

>> cat urls.txt
https://regex101.com/
https://docs.python.org/3/this-url-will-404.html
https://www.nytimes.com/guides/
https://www.mediamatters.org/
https://1.1.1.1/
https://www.politico.com/tipsheets/morning-money
https://www.bloomberg.com/markets/economics
https://www.ietf.org/rfc/rfc2616.txt

请求应该使用单独session，以利用session内部连接池的复用优势。下面我们看看完整程序，接着逐步分析。

"""Asynchronously get links embedded in multiple pages' HMTL."""

import asyncio
import logging
import re
import sys
from typing import IO
import urllib.error
import urllib.parse

import aiofiles
import aiohttp
from aiohttp import ClientSession

logging.basicConfig(
    format="%(asctime)s %(levelname)s:%(name)s: %(message)s",
    level=logging.DEBUG,
    datefmt="%H:%M:%S",
    stream=sys.stderr,
)
logger = logging.getLogger("areq")
logging.getLogger("chardet.charsetprober").disabled = True

HREF_RE = re.compile(r'href="(.*?)"')

async def fetch_html(url: str, session: ClientSession, **kwargs) -> str:
    """GET request wrapper to fetch page HTML.

    kwargs are passed to `session.request()`.
    """

    resp = await session.request(method="GET", url=url, **kwargs)
    resp.raise_for_status()
    logger.info("Got response [%s] for URL: %s", resp.status, url)
    html = await resp.text()
    return html

async def parse(url: str, session: ClientSession, **kwargs) -> set:
    """Find HREFs in the HTML of `url`."""
    found = set()
    try:
        html = await fetch_html(url=url, session=session, **kwargs)
    except (
        aiohttp.ClientError,
        aiohttp.http_exceptions.HttpProcessingError,
    ) as e:
        logger.error(
            "aiohttp exception for %s [%s]: %s",
            url,
            getattr(e, "status", None),
            getattr(e, "message", None),
        )
        return found
    except Exception as e:
        logger.exception(
            "Non-aiohttp exception occured:  %s", getattr(e, "__dict__", {})
        )
        return found
    else:
        for link in HREF_RE.findall(html):
            try:
                abslink = urllib.parse.urljoin(url, link)
            except (urllib.error.URLError, ValueError):
                logger.exception("Error parsing URL: %s", link)
                pass
            else:
                found.add(abslink)
        logger.info("Found %d links for %s", len(found), url)
        return found

async def write_one(file: IO, url: str, **kwargs) -> None:
    """Write the found HREFs from `url` to `file`."""
    res = await parse(url=url, **kwargs)
    if not res:
        return None
    async with aiofiles.open(file, "a") as f:
        for p in res:
            await f.write(f"{url}\t{p}\n")
        logger.info("Wrote results for source URL: %s", url)

async def bulk_crawl_and_write(file: IO, urls: set, **kwargs) -> None:
    """Crawl & write concurrently to `file` for multiple `urls`."""
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            tasks.append(
                write_one(file=file, url=url, session=session, **kwargs)
            )
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    import pathlib
    import sys

    assert sys.version_info >= (3, 7), "Script requires Python 3.7+."
    here = pathlib.Path(__file__).parent

    with open(here.joinpath("urls.txt")) as infile:
        urls = set(map(str.strip, infile))

    outpath = here.joinpath("foundurls.txt")
    with open(outpath, "w") as outfile:
        outfile.write("source_url\tparsed_url\n")

    asyncio.run(bulk_crawl_and_write(file=outpath, urls=urls))

逐步分析：
HREF_RE是一个正则表达式，用以提取我们最终需要的HTML中的href标签

>> HREF_RE.search('Go to <a href="https://realpython.com/">Real Python</a>')
<re.Match object; span=(15, 45), match='href="https://realpython.com/"'>

协程fetch_html()是对GET请求的包裹(wrapper)用以执行请求并解码返回的HTML页面。它发出请求，等待响应，在非200的状态下抛出异常

resp = await session.request(method="GET", url=url, **kwargs)
resp.raise_for_status()

如果返回状态值适合，fetch_html()返回HTML页面(一个字符串)。在这个函数中没有异常处理，逻辑是将异常传递给调用者(caller)

html = await resp.text()

(2022.07.08 Fri)
我们await两个结果session.request()和resp.text()，他们都是可被等待的协程。请求-响应的循环应该(would otherwise be)是使用中长尾(long-tailed)和耗时(time-hogging)的部分，但因为异步IO的存在，fetch_html()让事件循环工作于其他稳定可行的job比如解析和写已经获得的URLs。

协程链的下一步是parse()，该方法给定URL，等待fetch_html()，并提取该HTML页面中所有href标签，确保所有标签都可用并形成一个绝对路径。

注意到parse()的第二部分是阻塞的，其中包含一个快速的正则匹配，以确保被发现的连接都被转换成路径。

在这个案例中，同步代码应该运行快且不太多和显著，但一个协程的任何一行，只要不含有yield、await或return，则这一行就有可能阻塞其他协程。如果解析这一步的工作量更大，可以考虑将这部分进程放在loop.run_in_executor()中运行，即放在进程池中。

下一步是协程的write()，这个方法等待parse()方法返回一系列经过解析的URLs，将返回结果用aiofiles包异步写进对应的文件。

最后一步，buld_crawl_and_write()作为协程链的入口函数(main entry point)，它使用单独session，对从urls.txt中读取的每个URL创建一个任务。

另有几点值得记录：

• 默认的ClientSession自带了适配器(adapter)，可处理100个开放连接。修改该参数可通过将asyncio.connector.TCPConnector的一个实例(instance)发送给ClientSession实现，也可以针对每个host设定上限(specify limits on a per-host basis)。
• 可对每个请求或总体请求的session设置最大timeouts
• 脚本中使用了异步前后文管理器async with命令，从同步前后文管理器到异步的转变非常直觉，后者中定义了__aenter__()和__aexit()__方法而非__exit__()和__enter__()方法。async with只能应用于被async def修饰的协程函数中。

脚本运行结果如下

$ python3 areq.py
21:33:22 DEBUG:asyncio: Using selector: KqueueSelector
21:33:22 INFO:areq: Got response [200] for URL: https://www.mediamatters.org/
21:33:22 INFO:areq: Found 115 links for https://www.mediamatters.org/
21:33:22 INFO:areq: Got response [200] for URL: https://www.nytimes.com/guides/
21:33:22 INFO:areq: Got response [200] for URL: https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Got response [200] for URL: https://www.ietf.org/rfc/rfc2616.txt
21:33:22 ERROR:areq: aiohttp exception for https://docs.python.org/3/this-url-will-404.html [404]: Not Found
21:33:22 INFO:areq: Found 120 links for https://www.nytimes.com/guides/
21:33:22 INFO:areq: Found 143 links for https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Wrote results for source URL: https://www.mediamatters.org/
21:33:22 INFO:areq: Found 0 links for https://www.ietf.org/rfc/rfc2616.txt
21:33:22 INFO:areq: Got response [200] for URL: https://1.1.1.1/
21:33:22 INFO:areq: Wrote results for source URL: https://www.nytimes.com/guides/
21:33:22 INFO:areq: Wrote results for source URL: https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Got response [200] for URL: https://www.bloomberg.com/markets/economics
21:33:22 INFO:areq: Found 3 links for https://www.bloomberg.com/markets/economics
21:33:22 INFO:areq: Wrote results for source URL: https://www.bloomberg.com/markets/economics
21:33:23 INFO:areq: Found 36 links for https://1.1.1.1/
21:33:23 INFO:areq: Got response [200] for URL: https://regex101.com/
21:33:23 INFO:areq: Found 23 links for https://regex101.com/
21:33:23 INFO:areq: Wrote results for source URL: https://regex101.com/
21:33:23 INFO:areq: Wrote results for source URL: https://1.1.1.1/

可计算行数

$ wc -l foundurls.txt
  626 foundurls.txt

异步IO的使用

这部分我们讨论何时使用异步IO。

何时、为什么选择异步IO

选择异步IO往往是以多线程和多进程作为参照物。事实上当你面临CPU密集型任务，比如sklearn或keras中的grid search，多进程明显是最优选项。

多线程和异步IO的多少有些类似。多线程的问题在于，哪怕其容易实现，有可能因为竞速问题(race condition)和内存使用等问题，导致无法跟踪的bug。

在扩展上，多线程没有异步IO优雅，因其实一个有限可用性的系统资源(system resource with a finit availablity)。在很多机器上创建几千个线程会导致系统崩溃但创建几千个协程完全可行。

异步IO在使用IO密集型任务时更加适合，比如

• 网络IO，不管你的程序运行在服务器端还是客户端
• 无服务器设计(serverless design)，比如端到端(peer-to-peer)，聊天室之类的多用户网络
• 当你想模拟fire-and-forget风格进行读写操作，并且不太担心对所操作的对象加锁的情况下

不使用异步IO的最大原因是await仅支持一些定义了特定方法的特定对象。如果想对DBMS系统做异步写操作，不仅需要找到该DBMS的Python wrapper，还要找到支持async/await语法的wrapper。包含同步调用的协程会阻塞其他协程和任务。

下面这些Python包支持异步操作：
From aio-libs

• aiohttp: Asynchronous HTTP client/server framework
• aioredis: Async IO Redis support
• aiopg: Async IO PostgreSQL support
• aiomcache: Async IO memcached client
• aiokafka: Async IO Kafka client
• aiozmq: Async IO ZeroMQ support
• aiojobs: Jobs scheduler for managing background tasks
• async_lru: Simple [LRU cache for async IO

From [magicstack:

• uvloop: Ultra fast async IO event loop
• asyncpg: (Also very fast) async IO PostgreSQL support

From other hosts:

• trio: Friendlier asyncio intended to showcase a radically simpler design
• aiofiles: Async file IO
• asks: Async requests-like http library
• asyncio-redis: Async IO Redis support
• aioprocessing: Integrates multiprocessing module with asyncio
• umongo: Async IO MongoDB client
• unsync: Unsynchronize asyncio
• aiostream: Like itertools, but async

选择了异步IO但用哪个包

除了asyncio还有trio和curio可选。

asyncio的杂七杂八(miscellaneous)

asyncio中的其他顶级函数

asyncio.create_task()和asyncio.gather()。
create_task()用于安排协程对象的执行，与asyncio.run()同用。

>> import asyncio

>> async def coro(seq) -> list:
...     """'IO' wait time is proportional to the max element."""
...     await asyncio.sleep(max(seq))
...     return list(reversed(seq))
...
>> async def main():
...     # This is a bit redundant in the case of one task
...     # We could use `await coro([3, 2, 1])` on its own
...     t = asyncio.create_task(coro([3, 2, 1]))  # Python 3.7+
...     await t
...     print(f't: type {type(t)}')
...     print(f't done: {t.done()}')
...
>> t = asyncio.run(main())
t: type <class '_asyncio.Task'>
t done: True

这个结构的精妙之处(subtlety)在于，如果在main()中不使用await t，可能在main()执行完成之前就停止。因为asyncio.run(main())调用了loop.run_until_complete(main())，没有await t的事件循环只关心是否main()执行完成，不关心main()中创建的任务是否完成。没有await t，循环中的其他任务可能会在执行前被取消。可使用asyncio.Task.all_tasks()查看当前pending的任务。

asyncio.create_taks()在Python 3.7引入，之前版本中请使用asuncio.ensure_future()代替。

关于asyncio.gather()，如果没有特殊情况，gather()用于将多个协程(futures)放进一个future，返回一个单一future对象，并且如果await asyncio.gather()且指定多个任务或协程，将等待所有任务的完成。gather()的返回结果是输入的返回结果列表(a list of the results across the inputs)。

>>> import time
>>> async def main():
...     t = asyncio.create_task(coro([3, 2, 1]))
...     t2 = asyncio.create_task(coro([10, 5, 0]))  # Python 3.7+
...     print('Start:', time.strftime('%X'))
...     a = await asyncio.gather(t, t2)
...     print('End:', time.strftime('%X'))  # Should be 10 seconds
...     print(f'Both tasks done: {all((t.done(), t2.done()))}')
...     return a
...
>>> a = asyncio.run(main())
Start: 16:20:11
End: 16:20:21
Both tasks done: True
>>> a
[[1, 2, 3], [0, 5, 10]]

可能会意识到gather()等待futures或携程中所有结果的完成。或者可以用asyncio.as_completed()执行循环，可得到按完成顺序执行的结果。该函数返回一个迭代器，用于yields tasks as they finish。下面的结果中coro([3, 2, 1])在coro([10, 5, 0])完成之前可用，在gather()中不是这样。

>>> async def main():
...     t = asyncio.create_task(coro([3, 2, 1]))
...     t2 = asyncio.create_task(coro([10, 5, 0]))
...     print('Start:', time.strftime('%X'))
...     for res in asyncio.as_completed((t, t2)):
...         compl = await res
...         print(f'res: {compl} completed at {time.strftime("%X")}')
...     print('End:', time.strftime('%X'))
...     print(f'Both tasks done: {all((t.done(), t2.done()))}')
...
>>> a = asyncio.run(main())
Start: 09:49:07
res: [1, 2, 3] completed at 09:49:10
res: [0, 5, 10] completed at 09:49:17
End: 09:49:17
Both tasks done: True

await的位次(precedence)

await和yield尽管行为相似，但是await的级别要远高于yield，because it is more tightly bound, there are a number of instances where you’d need parentheses in a yield from statement that are not required in an analogous await statement.

总结

读过本文，你已经了解了一下内容：

• 异步IO是与语言无关的模型，并提供了协程间间接通信的有效并发方式
• Python的新关键字async和await的使用说明，用于标记和定义协程
• Python的asyncio包，提供了运行和管理协程的API

Reference

1 realpython, async-io-python