jianfeng huang

Posted on Oct 2

手写一个 60 行事件循环：async/await 到底在"等"什么？

#python #async #asyncio #programming

第一次写 asyncio 的人，通常会被同一个现象绊一下：

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

hello()          # 什么也没打印

调用了，没有输出，也没有报错——只是静静地返回了一个 <coroutine object>。再加上 RuntimeWarning: coroutine 'hello' was never awaited。

这篇不解释概念，直接把事件循环写出来。你能手写一个 60 行的循环，就再也不会对 await 感到玄学。

一、第一步：函数是怎么"暂停"的

async def 不是魔法，它和生成器是同一套底层机制。先看生成器：

def g():
    print("step 1")
    x = yield "a"        # 挂起，把 "a" 交出去
    print(f"step 2, got {x}")
    yield "b"
    print("step 3, done")

it = g()
print(it.send(None))     # step 1 / a
print(it.send("hello"))  # step 2, got hello / b

两个动作要分清：

send(None)：让函数跑到下一个挂起点，返回它 yield 出来的东西。
send(value)：从挂起点恢复，并把 value 作为 yield 表达式的值送进去。

async def 生成的协程对象，接口几乎一样，只是它 send 出去的不是业务数据，而是一个"我在等的东西"（awaitable）。你可以自己验证：

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

c = hello()
c.send(None)     # 打印 hello，然后抛 StopIteration —— 因为它跑完了

所以第一个困惑解开了：async def 定义一个协程函数，调用它只是造出一个协程对象，一行代码都不会执行。必须有人 send 它才动。

那么 await 做了什么？它相当于说："接下来我要等的东西在这儿，请把控制权交回去；等它好了再 send 我。"

二、手写一个 60 行事件循环

现在我们要实现三样东西，才能让上面那句话成立：

一个 awaitable：知道自己什么时候"好了"，并能在好了之后通知别人。
一个 Task：把协程包起来，负责驱动它（send 它）并处理它交出来的东西。
一个 Event Loop：不断取出"已经好了"的任务，继续驱动它们。

先写最小的 awaitable——一个带定时器的"未来值"：

import time
from collections import deque

class Future:
    def __init__(self):
        self.done = False
        self.result = None
        self.callbacks = []

    def set_result(self, value):
        self.done = True
        self.result = value
        for cb in self.callbacks:
            cb(self)          # 通知所有等它的人

    def __await__(self):      # 让 await 能作用于它
        if not self.done:
            yield self        # 关键：把"自己"交出去，等外面叫醒
        return self.result

__await__ 里那个 yield self 是整篇文章的题眼：await 的本质，就是把"我在等谁"这件事暴露给事件循环，然后挂起。

接着是 Task，它负责推着协程往前走：

class Task:
    def __init__(self, coro, loop):
        self.coro = coro
        self.loop = loop
        self.step()               # 立刻开跑第一步

    def step(self, future=None, _=None):
        try:
            # 推进协程：它 yield 出什么，就是它在等什么
            awaited = self.coro.send(None if future is None else future.result)
        except StopIteration as done:
            self.loop.tasks_done.append(done.value)
            return
        # 在它等的那个 future 上挂个回调：好了就再推我一把
        awaited.callbacks.append(self.step)

最后是事件循环本体——这一版直接写死"所有等待都是定时器"，够看懂就行：

class Loop:
    def __init__(self):
        self.timers = []          # [(fire_at, future)]
        self.ready = deque()      # 已就绪、等着被推进的任务
        self.tasks_done = []

    def sleep(self, seconds):
        fut = Future()
        self.timers.append((time.monotonic() + seconds, fut))
        return fut

    def run(self):
        while self.timers or self.ready:
            if not self.ready:
                # 没有就绪任务时，让出 CPU 等最近的一个定时器
                self.timers.sort()
                fire_at, fut = self.timers.pop(0)
                wait = fire_at - time.monotonic()
                if wait > 0:
                    time.sleep(wait)
                fut.set_result(None)      # 叫醒等它的人
            while self.ready:
                self.ready.popleft()()
        return self.tasks_done

Task.step 里那句 awaited.callbacks.append(self.step) 把两部分接起来了：协程挂起时说"我在等 fut"，循环在 fut 好了之后回调 Task.step，协程就从挂起点继续。

跑一个试试：

loop = Loop()

async def worker(name, delay):
    print(f"{name} 开始等")
    await loop.sleep(delay)
    print(f"{name} 等到了")
    return name

async def main():
    t1 = Task(worker("A", 1.0), loop)
    t2 = Task(worker("B", 0.5), loop)
    return "main done"

Task(main(), loop)
loop.run()

输出顺序是：A 开始等 → B 开始等 → B 等到了 → A 等到了。整个过程只有主线程，time.sleep 只出现在"所有协程都在等、CPU 确实没活干"的那一刻——这就是事件循环唯一允许阻塞的位置。

三、对照真正的 asyncio：名字换了，骨架一样

现在把上面的玩具和 CPython 里的真家伙对齐：

玩具版	真实 asyncio	职责
`Future`	`asyncio.Future`	一个"将来会有结果"的占位符，可挂回调、可 `await`
`Task`	`asyncio.Task`	驱动协程、保存状态、抛异常、存返回值
`Loop.run`	`loop.run_forever()` / `run_until_complete()`	从就绪队列取任务、用 selector 等 IO、触发定时器
`loop.sleep`	`asyncio.sleep`	挂起当前任务，登记定时器
手写 `callbacks`	`loop.call_soon` / `add_done_callback`	回调调度
`time.sleep(wait)`	`selectors.select(timeout)`	真正等 IO 就绪的地方

差别主要在最后一行：真实的事件循环等的不是定时器，而是 select/epoll/kqueue（由 selectors 模块抽象）上的文件描述符就绪事件。所以它同时能干三件事——socket 可读可写、定时器到期、别的线程 call_soon_threadsafe 丢进来的回调。

还有几个差异值得知道：

asyncio.run() 做了很多事：新建事件循环、把 main() 包成 Task、跑到完成、取消剩余任务、关掉 loop 和 executor。所以它全局只能调用一次，且不能嵌套。
异常传播：真实 Task 会把协程抛出的异常存进 Task 对象，等你 await 它时再抛出。这就是那句 Task exception was never retrieved 的来源——你没 await，异常就一直躺在那里。
取消：task.cancel() 其实是在协程当前挂起点扔一个 CancelledError。所以 except Exception 里包住 await 是常见 bug 来源——顺手把取消也吞了，应该用 except asyncio.CancelledError: raise 放行。
线程：await 时必须持有事件循环（asyncio.get_running_loop() 断言了这点），这就是为什么在普通同步函数里写 await 会直接 SyntaxError。

四、三个高频坑，现在应该能自己解释了

1. async def 里写 time.sleep(3) 为什么灾难？
因为它不 yield。time.sleep 在 C 层真的把线程睡住了，事件循环根本没机会跑回自己的主循环去推进其他任务。换成 await asyncio.sleep(3)，协程才会在挂起点把控制权交回去。

2. 为什么 async 会"传染"？
因为 await 只能出现在协程里，而一个协程要被真正驱动，必须有个 Task 在它上面跑 send。同步函数没法"等一半"——它没有挂起点。所以调用链上每一层都得变成协程，最终由事件循环收口。

3. 为什么 CPU 密集任务会拖垮整个服务？
事件循环是单线程的。一个任务在算数而不让出，等价于玩具版里某个 Task 的 step 里塞了个死循环——就绪队列里的其他任务永远轮不上，连 IO 就绪事件也没人去 select。这时候正确的动作不是"多开几个协程"，而是 await asyncio.to_thread(...) 或丢给进程池。

五、收口

一句话总结 await：它不是"我要结果"，而是"我在等谁，先让我下来，好了叫我"。

如果你想验证自己真懂了，做一件事就够了：把本文第二节那段代码抄进一个文件跑一遍，然后故意把 Task.step 里的回调注册删掉——你会看到程序卡死或者什么都不发生。那一刻你就摸到"挂起"和"被唤醒"之间的那根线了，而这根线，就是协程的全部秘密。

DEV Community