雾里看花之 Python Asyncio

从 Twisted 框架借鉴一些经验来理解 asynio 并非难事，但是，asyncio 包含众多的元素，我开始动摇，不知道如何将这些孤立的零碎拼图组合成一副完整的图画。我已没有足够的智力提出任何更好的建议，在这里，只想分享我的困惑，求大神指点。

原语

asyncio 通过协程coroutines 的帮助来实现异步 IO。最初它是通过 yield 和 yield from 表达式实现的一个库，因为 Python 语言本身演进的缘故，现在它已经变成一个更复杂的怪兽。所以，为了在同一个频道讨论下去，你需要了解如下一些术语：

• 事件循环
• 事件循环策略
• awaitable
• 协程函数
• 老式协程函数
• 协程
• 协程封装
• 生成器generator
• future
• 并发的future
• 任务task
• 句柄
• 执行器executor
• 传输transport
• 协议

此外，Python 还新增了一些新的特殊方法：

• __aenter__ 和 __aenter__，用于异步块操作
• __aiter__ 和 __anext__，用于异步迭代器（异步循环和异步推导）。为了更强大些，协议已经改变过一次了。 在 Python 3.5 它返回一个 awaitable（这是个协程）；在 3.6它返回一个新的异步生成器。
• __await__，用于自定义的 awaitable

你还需要了解相当多的内容，文档涵盖了那些部分。尽管如此，我做了一些额外说明以便对其有更好的理解：

事件循环

asyncio 事件循环和你第一眼看上去的略有不同。表面看，每个线程都有一个事件循环，然而事实并非如此。我认为它们应该按照如下的方式工作：

• 如果是主线程，当调用 asyncio.get_event_loop() 时创建一个事件循环。
• 如果是其它线程，当调用 asyncio.get_event_loop() 时返回运行时错误。
• 当前线程可以使用 asyncio.set_event_loop() 在任何时间节点绑定事件循环。该事件循环可由 asyncio.new_evet_loop() 函数创建。
• 事件循环可以在不绑定到当前线程的情况下使用。
• asyncio.get_event_loop() 返回绑定线程的事件循环，而非当前运行的事件循环。

这些行为的组合是超混淆的，主要有以下几个原因。 首先，你需要知道这些函数被委托到全局设置的底层事件循环策略。 默认是将事件循环绑定到线程。 或者，如果需要的话，可以在理论上将事件循环绑定到一个 greenlet 或类似的。 然而，重要的是要知道库代码不控制策略，因此不能推断 asyncio 将适用于线程。

其次，asyncio 不需要通过策略将事件循环绑定到上下文。 事件循环可以单独工作。 但是这正是库代码的第一个问题，因为协同程序或类似的东西并不知道哪个事件循环负责调度它。 这意味着，如果从协程中调用 asyncio.get_event_loop()，你可能没有机会取得事件循环。 这也是所有 API 均采用可选的显式事件循环参数的原因。 举例来说，要弄清楚当前哪个协程正在运行，不能使用如下调用：

1. def get_task():
2. loop = asyncio.get_event_loop()
3. try:
4. return asyncio.Task.get_current(loop)
5. except RuntimeError:
6. return None

相反，必须显式地传递事件循环。 这进一步要求你在库代码中显式地遍历事件循环，否则可能发生很奇怪的事情。 我不知道这种设计的思想是什么，但如果不解决这个问题（例如 get_event_loop() 返回实际运行的事件循环），那么唯一有意义的其它方案是明确禁止显式事件循环传递，并要求它绑定到当前上下文（线程等）。

由于事件循环策略不提供当前上下文的标识符，因此库也不可能以任何方式“索引”到当前上下文。 也没有回调函数用来监视这样的上下文的拆除，这进一步限制了实际可以开展的操作。

awaitable 与协程coroutine

以我的愚见，Python 最大的设计错误是过度重载迭代器。它们现在不仅用于迭代，而且用于各种类型的协程。 Python 中迭代器最大的设计错误之一是如果 StopIteration 没有被捕获形成的空泡。 这可能导致非常令人沮丧的问题，其中某处的异常可能导致其它地方的生成器或协同程序中止。 这是一个长期存在的问题，基于 Python 的模板引擎如 Jinja 经常面临这种问题。 该模板引擎在内部渲染为生成器，并且当由于某种原因的模板引起 StopIteration 时，渲染就停止在那里。

Python 慢慢认识到了过度重载的教训。 首先在 3.x 版本加入 asyncio 模块，并没有语言级支持。 所以自始至终它不过仅仅是装饰器和生成器而已。 为了实现 yield from 以及其它东西，StopIteration再次重载。 这导致了令人困惑的行为，像这样：

1. >>> def foo(n):
2. ... if n in (0, 1):
3. ... return [1]
4. ... for item in range(n):
5. ... yield item * 2
6. ...
7. >>> list(foo(0))
8. []
9. >>> list(foo(1))
10. []
11. >>> list(foo(2))
12. [0, 2]

没有错误，没有警告。只是不是你所期望的行为。 这是因为从一个作为生成器的函数中 return 的值实际上引发了一个带有单个参数的 StopIteration，它不是由迭代器协议捕获的，而只是在协程代码中处理。

在 3.5 和 3.6 有很多改变，因为现在除了生成器我们还有协程对象。除了通过封装生成器来生成协程，没有其它可以直接生成协程的单独对象。它是通过用给函数加 async 前缀来实现。 例如 async def x()会产生这样的协程。 现在在 3.6，将有单独的异步生成器，它通过触发 AsyncStopIteration 保持其独立性。 此外，对于Python 3.5 和更高版本，导入新的 future 对象（generator_stop），如果代码在迭代步骤中触发 StopIteration，它将引发 RuntimeError。

为什么我提到这一切？ 因为老的实现方式并未真的消失。 生成器仍然具有 send 和 throw 方法以及协程仍然在很大程度上表现为生成器。你需要知道这些东西，它们将在未来伴随你相当长的时间。

为了统一很多这样的重复，现在我们在 Python 中有更多的概念了：

• awaitable：具有__await__方法的对象。 由本地协同程序和旧式协同程序以及一些其它程序实现。
• 协程函数coroutinefunction：返回原生协程的函数。 不要与返回协程的函数混淆。
• 协程coroutine: 原生的协程程序。 注意，目前为止，当前文档不认为老式 asyncio 协程是协程程序。 至少 inspect.iscoroutine 不认为它是协程。 尽管它被 future/awaitable 分支接纳。

特别令人困惑的是 asyncio.iscoroutinefunction 和inspect.iscoroutinefunction 正在做不同的事情，这与 inspect.iscoroutine 和 inspect.iscoroutinefunction 情况相同。 值得注意的是，尽管inspect 在类型检查中不知道有关 asycnio 旧式协程函数的任何信息，但是当您检查 awaitable 状态时它显然知道它们，即使它与 await 不一致。

协程封装器coroutine wrapper

每当你运行 async def ，Python 就会调用一个线程局部的协程封装器。它由sys.set_coroutine_wrapper 设置，并且它是可以包装这些东西的一个函数。 看起来有点像如下代码：

1. >>> import sys
2. >>> sys.set_coroutine_wrapper(lambda x: 42)
3. >>> async def foo():
4. ... pass
5. ...
6. >>> foo()
7. __main__:1: RuntimeWarning: coroutine 'foo' was never awaited
8. 42

在这种情况下，我从来没有实际调用原始的函数，只是给你一个提示，说明这个函数可以做什么。 目前我只能说它总是线程局部有效，所以，如果替换事件循环策略，你需要搞清楚如何让协程封装器在相同的上下文同步更新。创建的新线程不会从父线程继承那些标识。

这不要与 asyncio 协程封装代码混淆。

awaitable 和 future

有些东西是 awaitable 的。 据我所见，以下概念被认为是 awaitable：

• 原生的协程
• 配置了假的 CO_ITERABLE_COROUTINE 标识的生成器（文中有涉及）
• 具有 __await__ 方法的对象

除了生成器由于历史遗留的原因不使用之外，其它的对象都使用 __await__ 方法。CO_ITERABLE_COROUTINE 标志来自哪里？它来自一个协程封装器（现在与sys.set_coroutine_wrapper 有些混淆），即 @asyncio.coroutine。 通过一些间接方法，它使用types.coroutine（现在与 types.CoroutineType 或 asyncio.coroutine 有些混淆）封装生成器，并通过另外一个标志 CO_ITERABLE_COROUTINE 重新创建内部代码对象。

所以既然我们知道这些东西是什么，那么什么是 future？ 首先，我们需要澄清一件事情：在 Python 3 中，实际上有两种（完全不兼容）的 future 类型：asyncio.futures.Future 和concurrent.futures.Future。 其中一个出现在另一个之前，但它们都仍然在 asyncio 中使用。 例如，asyncio.run_coroutine_threadsafe() 将调度一个协程到在另一个线程中运行的事件循环，但它返回一个 concurrent.futures.Future 对象，而不是 asyncio.futures.Future 对象。 这是有道理的，因为只有 concurrent.futures.Future 对象是线程安全的。

所以现在我们知道有两个不兼容的 future，我们应该澄清哪个 future 在 asyncio 中。 老实说，我不完全确定差异在哪里，但我打算暂时称之为“最终”。它是一个最终将持有一个值的对象，当还在计算时你可以对最终结果做一些处理。 future 对象的一些变种称为 deferred，还有一些叫做 promise。 我实在难以理解它们真正的区别。

你能用一个 future 对象做什么？ 你可以关联一个准备就绪时将被调用的回调函数，或者你可以关联一个 future 失败时将被触发的回调函数。 此外，你可以 await 它（它实现__await__，因此可等待），此外，future 也可以取消。

那么你怎样才能得到这样的 future 对象？ 通过在 awaitable 对象上调用 asyncio.ensure_future。它会把一个旧版的生成器转变为 future 对象。 然而，如果你阅读文档，你会读到asyncio.ensure_future 实际上返回一个task（任务）。 那么问题来了，什么是任务？

任务

任务task某种意义上是一个封装了协程的 futur 对象。它的工作方式和 future 类似，但它也有一些额外的方法来提取所包含的协程的当前堆栈。 我们已经见过了在前面提到过的任务，因为它是通过Task.get_current 确定事件循环当前正在做什么的主要方式。

在如何取消工作方面，任务和 future 也有区别，但这超出了本文的范围。“取消”是它们自己最大的问题。 如果你处于一个协程中，并且知道自己正在运行，你可以通过前面提到的 Task.get_current 获取自己的任务，但这需要你知道自己被派遣在哪个事件循环，该事件循环可能是、也可能不是已绑定的那个线程。

协程不可能知道它与哪个循环一起使用。task 也没有提供该信息的公共 API。 然而，如果你确实可以获得一个任务，你可以访问 task._loop，通过它反指到事件循环。

句柄

除了上面提到的所有一切还有句柄。 句柄是等待执行的不透明对象，不可等待，但可以被取消。 特别是如果你使用 call_soon 或者 call_soon_threadsafe（还有其它一些）调度执行一个调用，你可以获得句柄，然后使用它尽力尝试取消执行，但不能等待实际调用生效。

执行器Executor

因为你可以有多个事件循环，但这并不意味着每个线程理所当然地应用多个事件循环，最常见的情形还是一个线程一个事件循环。 那么你如何通知另一个事件循环做一些工作？ 你不能到另一个线程的事件循环中执行回调函数并获取结果。 这种情况下，你需要使用执行器。

执行器Executor来自 concurrent.futures，它允许你将工作安排到本身未发生事件的线程中。 例如，如果在事件循环中使用 run_in_executor 来调度将在另一个线程中调用的函数。 其返回结果是 asyncio 协程，而不是像 run_coroutine_threadsafe 这样的并发协程。 我还没有足够的心智来弄清楚为什么设计这样的 API，应该如何使用，以及什么时候使用。 文档中建议执行器可以用于构建多进程。

传输和协议

我总是认为传输与协议也凌乱不堪，实际这部分内容基本上是对 Twisted 的逐字拷贝。详情毋庸赘述，请直接阅读相关文档。

如何使用 asyncio

现在我们已经大致了解 asyncio，我发现了一些模式，人们似乎在写 asyncio 代码时使用：

• 将事件循环传递给所有协程。 这似乎是社区中一部分人的做法。 把事件循环信息提供给协程为协程获取自己运行的任务提供了可能性。
• 或者你要求事件循环绑定到线程，这也能达到同样的目的。 理想情况下两者都支持。 可悲的是，社区已经分化。
• 如果想使用上下文数据（如线程本地数据），你可谓是运气不佳。 最流行的变通方法显然是 atlassian 的 aiolocals，它基本上需要你手动传递上下文信息到协程，因为解释器不为此提供支持。 这意味着如果你用一个工具类库生成协程，你将失去上下文。
• 忽略 Python 中的旧式协程。 只使用 3.5 版本中 async def 关键字和协程。 你总可能要用到它们，因为在老版本中，没有异步上下文管理器，这是非常必要的资源管理。
• 学习重新启动事件循环进行善后清理。 这部分功能和我预想的不同，我花了比较长的时间来厘清它的实现。清理操作的最好方式是不断重启事件循环直到没有等待事件。 遗憾的是没有什么通用的模式来处理清理操作，你只能用一些丑陋的临时方案糊口度日。 例如 aiohttp 的 web 支持也做这个模式，所以如果你想要结合两个清理逻辑，你可能需要重新实现它提供的工具助手，因为该助手功能实现后，它彻底破坏了事件循环的设计。 当然，它不是我见过的第一个干这种坏事的库 :(。
• 使用子进程是不明显的。 你需要一个事件循环在主线程中运行，我想它是在监听信号事件，然后分派到其它事件循环。 这需要通过 asyncio.get_child_watcher().attach_loop(...) 通知循环。
• 编写同时支持异步和同步的代码在某种程度上注定要失败。 尝试在同一个对象上支持 with 和async with 是危险的事情。
• 如果你想给一个协程起个更好的名字，弄清楚为什么它没有被等待，设置 __name__没有帮助。 你需要设置 __qualname__ 而不是打印出错误消息来。
• 有时内部类型交换会使你麻痹。 特别是 asyncio.wait() 函数将确保所有的事情都是 future，这意味着如果你传递协程，你将很难发现你的协程是否已经完成或者正在等待，因为输入对象不再匹配输出对象。 在这种情况下，唯一真正理智的做法是确保前期一切都是 future。

上下文数据

除了疯狂的复杂性和对如何更好地编写 API 缺乏理解，我最大的问题是完全缺乏对上下文本地数据的考虑。这是 Node 社区现在学习的东西。continuation-local-storage 存在，但该实现被接受的太晚。持续本地存储和类似的概念常用于在并发环境中实施安全策略，并且该信息的损坏可能导致严重的安全问题。

事实上，Python 甚至没有任何存储，这令人失望至极。我正在研究这个内容，因为我正在调查如何最好地支持 Sentry's breadcrumbs 的 asyncio，然而我并没有看到一个合理的方式做到这一点。在 asyncio 中没有上下文的概念，没有办法从通用代码中找出您正在使用的事件循环，并且如果没有 monkeypatching（运行环境下的补丁），也无法获取这些信息。

Node 当前正在经历如何找到这个问题的长期解决方案的过程。这个问题不容忽视，因为它在所有生态系统中反复出现过，如 JavaScript、Python 和 .NET 环境。该问题被命名为异步上下文传播，其解决方案有许多名称。在 Go 中，需要使用上下文包，并明确地传递给所有 goroutine（不是一个完美的解决方案，但至少有一个）。.NET 具有本地调用上下文形式的最佳解决方案。它可以是线程上下文，Web 请求上下文或类似的东西，除非被抑制，否则它会自动传播。微软的解决方案是我们的黄金标准。我现在相信，微软在 15 年前已经解决了该问题。

我不知道该生态系统是否还够年轻，还可以添加逻辑调用上下文，可能现在仍然为时未晚。

原文发布时间为：2017-12-21

本文来自云栖社区合作伙伴“Linux中国”