CPython asyncio 概念全景：事件循环、协程、Task 与 await 的底层机制详解

本文为 CPython 官方文档 Doc/howto/a-conceptual-overview-of-asyncio.rst 的深化解读，围绕 asyncio 的核心组件——事件循环、协程函数、协程对象、Task、await 与 Future——建立一套稳固的心智模型。读完本文，你将能回答三个关键问题：await 一个对象时幕后究竟发生了什么；asyncio 如何区分"不占 CPU 的任务"（如网络请求）与"需要 CPU 的任务"（如计算阶乘）；以及如何用 Future 亲手实现一个 asyncio.sleep。文中所有源码证据均取自当前仓库的 Lib/asyncio 标准库实现。

第一部分：高层概念

事件循环：一切皆相对它发生

asyncio 中的一切都相对于事件循环（event loop）发生。官方文档将其比喻为"乐队指挥"：它拥有一些被明确授予的权力，但更多能力来自"乐队成员"（各个任务）的配合与让渡。

更技术地说，事件循环内部维护着一个待执行的作业（jobs）集合。有些作业由你直接添加，有些则由 asyncio 间接添加。事件循环从待办列表中取出一个作业并"调用"它（即交出控制权），作业运行；当它暂停或完成时，控制权归还事件循环，事件循环再从池中挑选下一个作业。你可以粗略地把作业集合想成一条队列：作业被加入、然后逐个处理（大体按顺序，但不总是如此）。这个循环无限重复；若没有待执行作业，事件循环会"休眠"以避免空耗 CPU，直到 I/O 完成或定时器到期才再次被唤醒。

import asyncio

# 创建一个事件循环，并让它无限循环处理作业
event_loop = asyncio.new_event_loop()
event_loop.run_forever()

协作式调度有一个隐含前提：作业必须"懂得分享"。一个贪婪的作业可以独占控制权，让其他作业挨饿——这正是后文 await coroutine 陷阱的根源。

异步函数与协程对象

普通的 Python 函数调用即执行函数体：

def hello_printer():
    print(
        "Hi, I am a lowly, simple printer, though I have all I "
        "need in life -- \nfresh paper and my dearly beloved octopus "
        "partner in crime."
    )

而 async def（而不是普通 def）定义的是异步函数（协程函数，coroutine function）——调用它并不会执行函数体，而是创建并返回一个协程对象（coroutine object）：

async def loudmouth_penguin(magic_number: int):
    print(
        "I am a super special talking penguin. Far cooler than that printer. "
        f"By the way, my lucky number is: {magic_number}."
    )

>>> loudmouth_penguin(magic_number=3)
<coroutine object loudmouth_penguin at 0x104ed2740>

术语辨析很关键："协程函数"与"协程对象"常被混称为 coroutine。在本文语境中，coroutine 特指协程对象，即 types.CoroutineType 的原生协程实例；注意协程也可以是 collections.abc.Coroutine 的实例——这一区分对类型检查（type checking）有实际意义。

协程代表函数的函数体/逻辑，它必须被显式启动（仅仅创建并不会启动它）。协程可以在函数体中的各个点暂停与恢复，这种"可暂停、可恢复"的能力正是异步行为的基石。

协程是构建在生成器（generator）之上的：生成器函数是包含 yield 的函数：

def get_random_number():
    # 这可算不上好的随机数生成器！
    print("Hi")
    yield 1
    print("Hello")
    yield 7
    print("Howdy")
    yield 4
    ...

与协程函数类似，调用生成器函数不会执行它，而是创建生成器对象。用内建函数 next 可以把生成器推进到下一个 yield——即"运行，然后暂停"：

>>> generator = get_random_number()
>>> next(generator)
Hi
1
>>> next(generator)
Hello
7

源码印证：asyncio 正是利用生成器的 send/throw 协议驱动协程的。在 Lib/asyncio/tasks.py 的 Task.__step_run_and_handle_result 中（约 L283-L291），任务每前进一步都是通过 coro.send(None) 或 coro.throw(exc) 来驱动协程的——"我们直接使用 send 方法，因为协程没有 __iter__ 和 __next__ 方法"（源码注释原话）。

Task：绑定到事件循环的协程

粗略地说，Task 是绑定到事件循环的协程（注意：是协程对象，不是协程函数）。创建 Task 会自动将其调度执行——本质上是向事件循环的待办列表中添加一个"运行它"的回调。推荐通过 asyncio.create_task 创建：

coroutine = loudmouth_penguin(magic_number=5)
# 创建 Task 对象，并通过事件循环调度其执行
task = asyncio.create_task(coroutine)

asyncio 会自动把 Task 与当前事件循环关联，这是刻意设计的简化：否则你得手工追踪事件循环对象，并把它传递给每一个想创建任务的协程函数。

源码印证：asyncio.create_task 本身只是一层薄封装，见 Lib/asyncio/tasks.py：

def create_task(coro, **kwargs):
    """Schedule the execution of a coroutine object in a spawn task.

    Return a Task object.
    """
    loop = events.get_running_loop()
    return loop.create_task(coro, **kwargs)

而真正的调度发生在 Task.__init__ 中（Lib/asyncio/tasks.py）：除非 eager_start 且事件循环正在运行（此时立即执行 __eager_start()），否则执行 self._loop.call_soon(self.__step, context=self._context)——Task 自身并不会被加入事件循环，被加入的只是指向 __step 的回调。

用 asyncio.run 管理事件循环

实践中，推荐使用 asyncio.run，它负责管理事件循环并确保给定的协程在程序继续前完成：

import asyncio

async def main():
    # 进行各种奇奇怪怪的异步操作……
    ...

if __name__ == "__main__":
    asyncio.run(main())
    # 在协程 main() 完成之前，程序不会到达下面的 print
    print("coroutine main() is done!")

源码印证：asyncio.run 底层是 Lib/asyncio/runners.py 中的 Runner 上下文管理器。Runner.__init__ 接受 debug 与 loop_factory 参数；Runner.run() 会把传入协程包装成任务（self._loop.create_task(coro, context=context)），并在退出时执行收尾——取消所有挂起任务、关闭异步生成器、关闭默认线程池执行器（见 Lib/asyncio/runners.py 的 close() 方法）。

Task 被垃圾回收的隐患

由于被加入事件循环的只是回调而非 Task 对象本身，如果 Task 对象在被事件循环调用前被垃圾回收，就可能出问题：

async def hello():
    print("hello!")

async def main():
    asyncio.create_task(hello())
    # 其他运行一段时间并把控制权交还给事件循环的异步指令……
    ...

asyncio.run(main())

由于第 5 行创建的 task 对象没有被任何引用持有，它可能在事件循环调用它之前被 GC 掉。当事件循环最终尝试运行该任务时，可能发现对象已不存在。另一种情形是：协程持有 task 引用，但协程本身先于 task 完成——协程退出后局部变量离开作用域，同样可能被回收。实际上 asyncio 与 Python GC 付出了相当努力来避免这种事（比如 Lib/asyncio/tasks.py 中 Task.__del__ 会在任务仍为 pending 时被销毁时记录 "Task was destroyed but it is pending!"），但那不是肆意冒险的理由——请始终持有 Task 的强引用（如存入列表）。

await：行为取决于对象类型

await 关键字常见于两种用法：

await task
await coroutine

关键在于：await 的行为取决于被 await 对象的类型。

await 一个 Task：把控制权交还给事件循环

async def plant_a_tree():
    dig_the_hole_task = asyncio.create_task(dig_the_hole())
    await dig_the_hole_task

    # 其他与种树相关的指令。
    ...

设想事件循环把控制权交给了 plant_a_tree() 的开头。协程创建了一个 task 并 await 它。await dig_the_hole_task 会做两件事：

1. 把一个回调（用于恢复 plant_a_tree()）加入 dig_the_hole_task 的回调列表；
2. 把控制权交还给事件循环。

稍后事件循环把控制权交给 dig_the_hole_task，任务完成它要做的事；任务结束后，把它的各个回调加入事件循环——在本案中就是"恢复 plant_a_tree()"。概括来说：被 await 的 task 完成后，原来的任务/协程会被重新加入事件循环的待办列表等待恢复。这是一个基础而可靠的心智模型；实践中的控制权交接稍复杂一些，但差别不大。

源码印证：这条路径在 Lib/asyncio/tasks.py 中清晰可见——当 Task 一步执行后 yield 出一个 Future 时，会执行 result.add_done_callback(self.__wakeup, context=self._context)，然后设置 self._fut_waiter = result。被等待的 Future 完成后触发 __wakeup（L359 起），后者再调用 self.__step() 恢复任务。这与文档"await 向 task 的回调列表加入恢复回调"的描述一一对应。

await 一个协程：不会交还控制权

与 Task 不同，直接 await coroutine 不会把控制权交还给事件循环！ 它等效于调用一个普通同步函数。若先 asyncio.create_task(...) 包装再 await，才会让出控制权。看这个例子：

import asyncio

async def coro_a():
    print("I am coro_a(). Hi!")

async def coro_b():
    print("I am coro_b(). I sure hope no one hogs the event loop...")

async def main():
    task_b = asyncio.create_task(coro_b())
    num_repeats = 3
    for _ in range(num_repeats):
        await coro_a()
    await task_b

asyncio.run(main())

main() 的第一句创建了 task_b 并调度执行。随后 coro_a() 被反复直接 await，控制权从未交给事件循环，所以三次 coro_a() 的输出全部排在 coro_b() 之前：

I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_b(). I sure hope no one hogs the event loop...

若把 await coro_a() 改为 await asyncio.create_task(coro_a())，行为就变了：main() 在该语句处让出控制权，事件循环先调用 task_b，再调用包装 coro_a() 的任务，然后恢复 main()：

I am coro_b(). I sure hope no one hogs the event loop...
I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_a(). Hi!

这个 await coroutine 的行为容易坑到很多人：它可能无意间从其他任务手中"霸占"控制权，事实上卡住事件循环。可以用 asyncio.run(..., debug=True) 开启调试模式来检测此类问题——它会记录任何独占执行超过 100 毫秒的协程（见官方文档 Doc/library/asyncio.rst 中的 debug mode 说明）。

这是一个刻意的设计取舍：用一点使用上的概念模糊，换性能。每次 await 一个 task，控制权都要一路上传到事件循环；事件循环又要处理内部状态、执行调度逻辑来恢复下一个作业。单个开销看似很小，但在有大量 await 的大程序中会累积成不可忽略的性能拖累。

第二部分：底层机制（nuts and bolts）

协程的内在工作方式

asyncio 利用 Python 的四个构件来回传递控制权：coroutine.send(arg)、yield、await（调用对象的 __await__ 方法）、以及 StopIteration。

• coroutine.send(arg) 用于启动或恢复协程。若协程是从暂停处恢复，arg 作为当初暂停它的 yield 语句的返回值被送入；若是首次使用（启动而非恢复），arg 必须是 None。

完整示例：

class Rock:
    def __await__(self):
        value_sent_in = yield 7
        print(f"Rock.__await__ resuming with value: {value_sent_in}.")
        return value_sent_in

async def main():
    print("Beginning coroutine main().")
    rock = Rock()
    print("Awaiting rock...")
    value_from_rock = await rock
    print(f"Coroutine received value: {value_from_rock} from rock.")
    return 23

coroutine = main()
intermediate_result = coroutine.send(None)
print(f"Coroutine paused and returned intermediate value: {intermediate_result}.")

print(f"Resuming coroutine and sending in value: 42.")
try:
    coroutine.send(42)
except StopIteration as e:
    returned_value = e.value
print(f"Coroutine main() finished and provided value: {returned_value}.")

逐步解析控制流与值的传递：

1. 第 16 行 coroutine.send(None) 首次启动协程，执行到第 11 行 await rock；
2. await 会调用对象的 __await__ 方法；而 Rock.__await__ 中的 yield 7（第 3 行）使协程暂停，值 7 沿调用链一路向上传播，回到第 16 行的调用处，成为 intermediate_result。await 还做了一件特别的事：它把收到的 yield 沿调用链继续传播（propagate）。

Beginning coroutine main().
Awaiting rock...
Coroutine paused and returned intermediate value: 7.
Resuming coroutine and sending in value: 42.
Rock.__await__ resuming with value: 42.
Coroutine received value: 42 from rock.
Coroutine main() finished and provided value: 23.

3. 第 21 行 coroutine.send(42) 恢复协程，它从第 3 行 yield 处继续，value_sent_in 即为 42；
4. 协程结束时抛出 StopIteration，返回值附在异常的 value 属性上（returned_value 即 main() 的返回值 23）。

值得注意的两个"为什么"：

1. 协程函数里直接 yield？ 那样它就成了异步生成器函数（async generator function），是完全不同的东西。
2. 协程函数里 yield from 一个普通生成器？ 会报 SyntaxError: yield from not allowed in a coroutine.。这是刻意为之——只保留使用协程的一条路径，换概念上的简洁。事实上 yield from 与 await 做的事基本相同。yield 最初也被禁止，后来为支持异步生成器才被重新允许。

因此，协程让出控制权（yield）的唯一方式，就是 await 一个其 __await__ 方法内含 yield 的对象。这一点在 C 层实现中同样成立：Task.__step_run_and_handle_result 中（Lib/asyncio/tasks.py）对裸 yield（yield 出 None）的处理是 self._loop.call_soon(self.__step, ...)，源码注释写明"裸 yield 让出控制权一个事件循环迭代"；yield 出 Future、生成器或其他值则分别触发等待逻辑或 RuntimeError。

源码印证（await Future 的完整闭环）：Lib/asyncio/futures.py 中 Future.__await__ 会先设置 self._asyncio_future_blocking = True 然后 yield self——这正对应文档描述的"__await__ 里 yield 即让出控制权"。随后 Task 侧检测到 _asyncio_future_blocking 并注册 __wakeup 回调（Lib/asyncio/tasks.py），Future 完成后 __wakeup 调用 __step() 恢复任务，恢复时 __step 再次调用 coro.send(None) 把控制权送回协程。文档中"await task 是向回调列表添加恢复回调"的高层描述，由此得到完整的底层印证。

Future：计算状态与结果的代表

Future（Doc/library/asyncio.rst 中的 asyncio-future-obj）表示一次计算的状态与结果——名字致敬"尚未到来之事"，对象就是盯住那件事的方式。

Future 的关键属性：

• 状态（state）：pending、cancelled 或 done 之一；
• 结果（result）：状态转为 done 时被设置。

与协程不同，Future 不代表要执行的实际计算，它代表的是该计算的状态与结果——好比交通信号灯（红、黄、绿）或指示器。

asyncio.Task 通过继承 asyncio.Future 获得这些能力。上一节说"task 存有一个回调列表"并不完全准确：真正实现回调逻辑的是 Future 类，Task 只是继承者（在 CPython 中为 class Task(futures._PyFuture)，见 Lib/asyncio/tasks.py，而 Future 的状态机定义在 Lib/asyncio/base_futures.py）。

Future 也可以脱离 Task 直接使用。Task 在其协程完成时把自己标记为 done；而 Future 更灵活——你说它 done 它才 done。这正是让你自定义"等待与恢复条件"的灵活接口。

源码印证：Future.set_result() 是"标记为完成"的入口（Lib/asyncio/futures.py）；Task 则禁止直接调用它（Lib/asyncio/tasks.py 中 Task.set_result 直接 raise RuntimeError('Task does not support set_result operation')）——Task 只能经由协程的 StopIteration 来完成。

亲手实现一个 asyncio.sleep

下面利用 Future 实现一个模拟 asyncio.sleep 的 async_sleep：注册若干任务到事件循环，然后 await 包装 async_sleep(3) 的 task，要求它在三秒后才完成，但不阻止其他任务运行。

async def other_work():
    print("I like work. Work work.")

async def main():
    # 向事件循环添加几个其他任务，这样异步睡眠时有事可做。
    work_tasks = [
        asyncio.create_task(other_work()),
        asyncio.create_task(other_work()),
        asyncio.create_task(other_work())
    ]
    print(
        "Beginning asynchronous sleep at time: "
        f"{datetime.datetime.now().strftime("%H:%M:%S")}."
    )
    await asyncio.create_task(async_sleep(3))
    print(
        "Done asynchronous sleep at time: "
        f"{datetime.datetime.now().strftime("%H:%M:%S")}."
    )
    # asyncio.gather 等效于 await 集合中的每个任务。
    await asyncio.gather(*work_tasks)

async_sleep 用一个 Future 来精确控制任务何时被标记为 done——如果 future.set_result()（负责把 Future 标记为 done 的方法）从未被调用，该任务永远不会结束：

async def async_sleep(seconds: float):
    future = asyncio.Future()
    time_to_wake = time.time() + seconds
    # 把观察者任务加入事件循环。
    watcher_task = asyncio.create_task(_sleep_watcher(future, time_to_wake))
    # 阻塞直到 future 被标记为 done。
    await future

再配合一个朴素的 YieldToEventLoop() 对象，从其 __await__ 方法中 yield，从而让出控制权。这等效于调用 asyncio.sleep(0)，但语义更清晰——毕竟用 asyncio.sleep 来演示如何实现 asyncio.sleep 有点作弊：

class YieldToEventLoop:
    def __await__(self):
        yield

async def _sleep_watcher(future, time_to_wake):
    while True:
        if time.time() >= time_to_wake:
            # 这把 future 标记为 done。
            future.set_result(None)
            break
        else:
            await YieldToEventLoop()

事件循环照常遍历任务：给予控制权，在它们暂停或完成时收回。watcher_task 每完整一轮事件循环被调用一次；每次恢复时检查时间，不够就再次暂停、归还控制权；时间一到，_sleep_watcher 标记 Future 为 done 并退出 while True。由于该辅助任务每轮只被调用一次，可以推断这个异步睡眠至少睡三秒而非恰好三秒——asyncio.sleep 亦然（官方实现基于 loop.call_later 定时器，见 Lib/asyncio/tasks.py：h = loop.call_later(delay, futures._set_result_unless_cancelled, future, result) 后 return await future）。

完整程序输出：

$ python custom-async-sleep.py
Beginning asynchronous sleep at time: 14:52:22.
I like work. Work work.
I like work. Work work.
I like work. Work work.
Done asynchronous sleep at time: 14:52:25.

这个实现"不必要地绕"了吗？是的。它的目的不是最优，而是用一个简单例子展示 Future 的通用性，且该模式可被仿照到更复杂的等待条件上。如果不需要"外部决定何时完成"的能力，可以不用 Future 直接写：

async def simpler_async_sleep(seconds):
    time_to_wake = time.time() + seconds
    while True:
        if time.time() >= time_to_wake:
            return
        else:
            await YieldToEventLoop()

（注意：simpler_async_sleep 中 await YieldToEventLoop() 是直接 await 协程外的 awaitable，每轮循环都会真正让出控制权，因此同样不会阻塞事件循环。）

心智模型小结

把两部分拼起来，asyncio 的控制权流转可以浓缩为一张表：

操作	是否让出控制权	底层机制（源码证据）
await task / await future	是	向 Future 添加 __wakeup 回调，控制权回到事件循环（Lib/asyncio/tasks.py）
await coroutine	否	等效于同步函数调用，直接驱动协程执行
create_task(coro)	否（调度是异步的）	loop.call_soon(self.__step) 入队回调（Lib/asyncio/tasks.py）
协程内 await YieldToEventLoop()	是	裸 yield 触发 call_soon(self.__step)（Lib/asyncio/tasks.py）
协程 return	—	抛出 StopIteration(value)，Task 借此 set_result（Lib/asyncio/tasks.py）

这套模型与 Doc/howto/a-conceptual-overview-of-asyncio.rst 的结论一致：理解"谁在什么时候把控制权交给谁"，是读懂 asyncio 推荐模式（如用 create_task + gather 并行、避免裸 await coroutine 阻塞循环、持有 Task 强引用、必要时开启 debug=True 检测独占执行）的全部前提。想进一步深入，可参阅 Doc/library/asyncio.rst 中 asyncio 完整 API 的其余章节。