异步任务的语法更完美

python编程语言对异步编程的支持一直在改进，比如python 2.0 版开始就增加了生成器(generator)，在3.4版开始增加了 asyncio 库，随后在 3.5 版中又增加了支持异步编程的关键字 async 和await。在最近发布的几个版本中，对异步编程能力也做了许多小的改进。在最新发布的 3.11 版中，可以使用任务组功能，语法更简洁，便于运行和监测异步任务。

asyncio 库是 python标准库的组成部分，但异步运行并不是其唯一的方式；另外，还有几个比较常用的、支持异步运行的第三方库，例如 Trio 和 Curio库，有些类似的第三方库如uvloop、AnyIO和Quatto在异步方面的能力比asyncio 更好，并且功能特点更多。

使用 asyncio 库执行多个异步任务的一般方法，先用 create_task() 创建任务，然后用 gether() 等待任务被执行完成，这种方式稍稍有些繁琐。而支持异步运行的第三方库提供了组织和管理子任务的功能，为此 Curio 引入了任务组(task group)的机制， Trio 引入了 nurseries 实现同样的功能。asyncio 作为python的标准库，也是受到了这种理念的启发，在异步编程方面引入了类似机制。

使用 asyncio 库进行异步编程时，用 gather() 组织 异步任务，代码可以这样写：

 tasks = [asyncio.create_task(run_some_task(param)) for param in params]
await asyncio.gather(*tasks)

在异步任务传递到 gather() 之前，可以人工跟踪所有的任务，等到 gather() 执行完成后，就可以确认每个任务都已经被执行完毕。上面的代码示例，用任务组写出来更加直观，可以不使用 gather() ，而代码所在环境下管理器来定义任务等待运行的时间，示例代码如下：

async with asyncio.TaskGroup() as tg:
    for param in params:
         tg.create_task(run_some_task(param))

以上示例代码中，创建了一个名称为 tg 的任务组对象，使用该对象的方法 .create_task() 来创建新的任务。

下面看一个示例，完成下载多个文件的任务，能充分体现 asyncio 的异步执行能力。这一示例中，下载python的官方历史文档PEP，该系列文档介绍了开发python的异步特色的过程，为了提高下载效率，使用了第三方库 aiohttp，采用异步下载的方式来完成，代码示例如下：

# download_peps_gather.py

import asyncio
import aiohttp

PEP_URL = (
    "https://raw.githubusercontent.com/python/peps/master/pep-{pep:04d}.txt"
)

async def main(peps):
    async with aiohttp.ClientSession() as session:
        await download_peps(session, peps)

以上代码中，增加了一个 main() 函数，用于初始化 aiohttp(), 以便管理可以重复使用的连接池； 调用下载任务函数 download_peps()，以上代码中还没有具体写出该函数的内容。该函数在下载每一个PEP文档时都会创建一个任务，download_peps_gather() 函数代码如下：

 # download_peps_gather.py

# ...

async def download_peps(session, peps):
    tasks = [asyncio.create_task(download_pep(session, pep)) for pep in peps]
    await asyncio.gather(*tasks)

以上代码的写法与前面的示例相同，每个任务都由运行download_pep()函数组成，并且确定了下一个任务，这就是前文所说的管理连接池的意义。创建了所有的任务之后，即可把这些任务传递给gather()；每个下载任务负责下载一个PEP文档，可以用print() 把下载文档的信息打印处理，以上代码改进如下：

 # download_peps_gather.py

# ...

async def download_pep(session, pep):
    print(f"Downloading PEP {pep}")
    url = PEP_URL.format(pep=pep)
    async with session.get(url, params={}) as response:
        pep_text = await response.text()

    title = pep_text.split("\n")[1].removeprefix("Title:").strip()
    print(f"Downloaded PEP {pep}: {title}")

以上代码中，对每一个要下载的PEP文档都有各自的URL，并且使用 session.get() 函数下载对应的文档。当PEP文档下载完，就可以得到PEP文档的名称，并且打印出来。现在，异步方式运行main()函数，下载指定编号的PEP文档，体验一下执行 python 异步代码的过程：

# download_peps_gather.py

# ...

asyncio.run(main([492, 525, 530, 3148, 3156]))

代码运行时，可以看到如下结果：

$ python download_peps_gather.py
Downloading PEP 492
Downloading PEP 525
Downloading PEP 530
Downloading PEP 3148
Downloading PEP 3156
Downloaded PEP 3148: futures - execute computations asynchronously
Downloaded PEP 492: Coroutines with async and await syntax
Downloaded PEP 530: Asynchronous Comprehensions
Downloaded PEP 3156: Asynchronous IO Support Rebooted: the "asyncio" Module
Downloaded PEP 525: Asynchronous Generators

可以看出，所有的下载同时进行，打印出的信息表明每个任务都开始下载，这里要注意启动下载任务的顺序是按照代码中定义的顺序进行的，这个例子中可以通过下载文档的编号看出来；但与之不同的是，下载任务的完成顺序看似是随机的，这里可以用gather() 方法来确认。修改前面的代码，使用任务组来替代 gather() 方法，并重新命名这段代码
download_peps_taskgroup.py, 修改前后的两段代码很相似；同时，download_peps() 函数也需要修改。修改后的代码如下：

# download_peps_taskgroup.py

# ...

async def download_peps(session, peps):
    async with asyncio.TaskGroup() as tg:
        for pep in peps:
            tg.create_task(download_pep(session, pep))

# ...

以上代码中，首先在需要任务组的地方创建一个任务组，然后用该任务组创建子任务，子任务就是下载每一个PEP文档；运行修改后的代码段，其结果与修改前的代码的运行结果一样。

如果代码中有多个异步任务之间协作，编写代码要费点精力，因为其中的任何一个异步任务都可能随时产生报错；理论上来说，两个或两个以上的异步任务同时报错是有可能的。对于这个问题的处理，第三方库 Trio 和 Curio 都是用了多个错误对象这样一种特殊的处理方式，这种方式虽然有效，但是有点繁琐，因为python3.11以前的版本并没有提供一种内在的机制支持这种方式。

python 3.11 为了处理多个异步任务之间协作时可能同时报错的情况，引入了异常处置组(exception error) 的方法，专门用于跟踪可能出现多个并发错误的情形。任务组(task group)使用异常处置组(exception group) 来处理相关报错，这种方法比 python 3.11版之前的方法更好。

（本文完。本系列文章根据官方正式发布的英文资料摘译、整理）

更多内容，可阅读本系列的其他文章：

python最新版3.11正式发布，有哪些新特色？(1/5)

python最新版3.11正式发布，有哪些新特色？(2/5)