Python 并发与 asyncio 14.1 高级 asyncio 协程与函数共存的 API

14 高级 asyncio

本章涵盖：

我们已经掌握了 asyncio 的绝大部分功能。利用前面章节介绍的 asyncio 模块，你应该能完成几乎任何你想要的任务。不过，还有一些不太为人知但可能会用到的技巧，特别是当你在设计自己的 asyncio API 时。

在本章中，我们将深入探讨 asyncio 中更高级的技术。我们会学习如何设计既能处理协程也能处理普通函数的 API，如何强制触发事件循环的迭代，以及如何在不传递参数的情况下在任务间传递状态。我们还会进一步挖掘 asyncio 如何利用生成器来实现异步操作，从底层理解其工作原理。通过实现自己的自定义可等待对象，并用它们构建一个能并发运行多个协程的简易事件循环，我们将获得深刻的领悟。

如果你不是在开发新的 API 或框架，而且平时编码也不太涉及异步编程的底层细节，那么本章的内容对你来说可能用处不大。这些技术主要适用于构建框架或对 Python 异步机制内部原理感兴趣的探索者。

14.1 协程与函数共存的 API

如果我们自己在设计一个库，可能不会假设用户一定是在异步应用中使用我们的库。他们可能还没迁移到异步模式，或者根本不需要从异步栈中获益，也许永远都不会迁移。那我们应该怎么设计一个兼容协程和普通函数的 API 呢？

asyncio 提供了几个方便的工具函数来帮助我们实现这一点：asyncio.iscoroutine 和 asyncio.iscoroutinefunction。这两个函数能帮我们判断一个可调用对象是否是协程，从而根据类型执行不同的逻辑。这些函数正是 Django 能无缝处理同步和异步视图的核心所在，我们在第 9 章里见识过。

来看看这个例子：我们来写一个任务运行器类，它能接收协程和普通函数。这个类会把任务加到内部列表里，当用户调用 start 方法时，会并发（如果任务是协程）或串行（如果任务是普通函数）地运行它们。

import asyncioclass TaskRunner:    def __init__(self):        self.loop = asyncio.new_event_loop()        self.tasks = []    def add_task(self, func):        self.tasks.append(func)    async def _run_all(self):        awaitable_tasks = []        for task in self.tasks:            if asyncio.iscoroutinefunction(task):                awaitable_tasks.append(asyncio.create_task(task()))            elif asyncio.iscoroutine(task):                awaitable_tasks.append(asyncio.create_task(task))            else:                self.loop.call_soon(task)        await asyncio.gather(*awaitable_tasks)    def run(self):        self.loop.run_until_complete(self._run_all())if __name__ == "__main__":    def regular_function():        print('Hello from a regular function!')    async def coroutine_function():        print('Running coroutine, sleeping!')        await asyncio.sleep(1)        print('Finished sleeping!')    runner = TaskRunner()    runner.add_task(coroutine_function)    runner.add_task(coroutine_function())    runner.add_task(regular_function)    runner.run()

在上面的例子中，我们的任务运行器创建了一个新的事件循环和一个空的任务列表。然后我们定义了 add 方法，用于把函数（或协程）加入待执行队列。当用户调用 run() 时，我们在事件循环里启动 _run_all 方法。这个方法会遍历所有任务，检查当前任务是否是协程。如果是，我们就创建一个任务；如果不是，就用事件循环的 call_soon 方法将普通的函数安排在下一轮循环中执行。最后，我们调用 gather 并等待所有任务完成。

接下来我们定义了两个函数：一个叫 regular_function，就是一个普通的函数；另一个是叫 coroutine_function，是一个协程。我们创建了 TaskRunner 的实例，并添加了三个任务。这里特意调用了两次 coroutine_function，来展示我们 API 中引用协程的两种不同方式。最终输出如下：

Running coroutine, sleeping!Running coroutine, sleeping!Hello from a regular function!Finished sleeping!Finished sleeping!

这说明我们成功同时运行了协程和普通函数。现在，我们已经构建了一个可以灵活处理协程和普通函数的 API，极大地丰富了终端用户的使用方式。接下来，我们看看上下文变量，它能让我们在不显式传参的情况下，为每个任务保存独立的状态。