更新页面 Python 并发语法
页面 kb/python-concurrency · 修改（rev 12） · · 2026-10-25 23:08
body
@@ -544,4 +544,180 @@ > [!say] Await only gives up control at the points I mark, so in asyncio a section without await is effectively atomic and needs no lock. Threads can be preempted anywhere, and you can't await a threading lock, so a multithreaded problem still needs a real lock. What does carry over from asyncio is confinement: let one thread own the state and have the others hand work to it. + +### 协程是怎么实现的 + +`asyncio` 的协程是三样东西拼起来的：Future、Task 和事件循环。下面先用一个小例子看协程怎么停下、怎么接着跑，再从零写一个 40 行的最小事件循环，逐步走一遍。看懂这一节，`await` 底下发生了什么就清楚了。 + +#### 协程怎么停下、怎么接着跑 + +> [!term] 生成器（generator） +> 函数体里有 `yield` 的函数。调用它不执行函数体，只得到一个生成器对象。每次调它的 `send()`，它就从上次停下的地方接着跑，跑到下一个 `yield` 停下，`yield` 出来的值就是 `send()` 的返回值。函数跑完时，`send()` 抛出 `StopIteration`。 + +协程对象也有 `send()`，用法和生成器一样。`await` 一个对象时，Python 会调用这个对象的 `__await__()`，它通常就是一个生成器，里面的 `yield` 就是协程的暂停点： + +```python + +class Box: # 一个最小的「可以被 await 的东西」 + def __init__(self): + self.done, self.value = False, None + def __await__(self): + if not self.done: + yield self # 暂停点：把自己交给外面调 send 的人 + return self.value # 恢复以后，这就是 await 表达式的值 + +box = Box() + +async def demo(): + print("1. 协程开始，跑到 await 之前") + x = await box + print("3. 协程恢复了，x =", x) + +coro = demo() # 只建了协程对象，函数体一行都没执行 +print("0. 建好了协程对象:", type(coro).__name__) +y = coro.send(None) # 跑到 yield 那里停下，yield 出来的东西就是 send 的返回值 +print("2. send 返回了:", "box" if y is box else y) +box.done, box.value = True, 42 # 相当于 set_result(42) +try: + coro.send(None) # 从 yield 下一行接着跑 +except StopIteration: + print("4. 协程跑完了，第二次 send 抛 StopIteration") +``` + +实测输出： + +``` +0. 建好了协程对象: coroutine +1. 协程开始，跑到 await 之前 +2. send 返回了: box +3. 协程恢复了，x = 42 +4. 协程跑完了，第二次 send 抛 StopIteration +``` + +逐行看： + +- 调 `demo()` 不执行函数体，只返回一个协程对象。第 0 行打印时，第 1 行还没打印。 +- 第一次 `send(None)` 从函数开头跑起，打印第 1 行，然后遇到 `await box`，进到 `box.__await__()` 里，执行到 `yield self` 停下。协程的局部变量和执行位置都保存在协程对象里，`yield` 出来的 `box` 成了 `send` 的返回值，所以第 2 行打印出 box。 +- 第二次 `send(None)` 从 `yield` 的下一行接着跑。这时 `box.done` 已经是 True，`__await__` 执行 `return self.value`，这个返回值就是 `await box` 整个表达式的值，所以 `x = 42`。函数跑完，`send` 抛 `StopIteration`。 + +协程自己没法恢复自己，谁调 `send`，谁就决定它什么时候接着跑。事件循环要做的，就是在合适的时候替它调 `send`。 + +#### 一个最小的事件循环 + +```python +import heapq, time +from collections import deque + +class Future: + """一个还没出来的结果。和 fire-event 的回调版一样：先登记回调，结果一出来就挨个执行。""" + def __init__(self, loop): + self.loop, self.done, self.result, self.callbacks = loop, False, None, [] + def add_done_callback(self, cb): + if self.done: + self.loop.call_soon(cb) + else: + self.callbacks.append(cb) + def set_result(self, value): # 相当于 fire + self.done, self.result = True, value + for cb in self.callbacks: + self.loop.call_soon(cb) + def __await__(self): + if not self.done: + yield self # 暂停：把自己交给 Task，意思是「等我好了再叫你」 + return self.result # 恢复以后，await 表达式的值就是结果 + +class Task: + """推着一个协程往前走：每次 send 让它跑到下一个 await。""" + def __init__(self, loop, coro): + self.loop, self.coro = loop, coro + loop.call_soon(self.step) + def step(self): + try: + fut = self.coro.send(None) # 跑到下一个 await 暂停，拿到它在等的 Future + except StopIteration: + return # 协程跑完了 + fut.add_done_callback(self.step) # Future 好了以后，再推一步 + +class Loop: + def __init__(self): + self.ready, self.timers = deque(), [] + def call_soon(self, cb): + self.ready.append(cb) + def call_at(self, when, cb): + heapq.heappush(self.timers, (when, id(cb), cb)) + def sleep(self, seconds): + fut = Future(self) + self.call_at(time.monotonic() + seconds, lambda: fut.set_result(None)) + return fut + def run(self): + while self.ready or self.timers: + while self.timers and self.timers[0][0] <= time.monotonic(): + self.ready.append(heapq.heappop(self.timers)[2]) + if not self.ready: # 没事可做：睡到最近的那个定时器 + time.sleep(max(0, self.timers[0][0] - time.monotonic())) + continue + self.ready.popleft()() + +loop = Loop() +log = [] +async def worker(name, delay): + log.append(f"{name} 开始") + await loop.sleep(delay) # 这里暂停，事件循环去跑别的 + log.append(f"{name} 结束") +t0 = time.perf_counter() +Task(loop, worker("A", 0.2)); Task(loop, worker("B", 0.1)) +loop.run() +print(log, f"{(time.perf_counter() - t0) * 1000:.0f} ms") +``` + +实测：日志是 A 开始、B 开始、B 结束、A 结束，总共约 200 毫秒，A 和 B 的等待叠在一起。三个类各干一件事： + +- **Future**：一个还没出来的结果，概念和上面线程池一节的 `Future` 一样。`add_done_callback(cb)` 登记回调，结果已经出来就直接把回调排进队列，没出来就先存着。`set_result(v)` 宣布结果，把存着的回调都排进队列。`__await__` 就是上一小节 `Box` 的那个写法。这两个方法和 [[q:everpure-fire-event]] 的回调版是一回事：`add_done_callback` 就是 `register`，`set_result` 就是 `fire`。 +- **Task**：推着一个协程往前走。`step` 调 `coro.send(None)`，让协程跑到下一个 `await` 停下，拿到它在等的 Future，再把自己的 `step` 登记成这个 Future 的回调，意思是「它出结果了，再来推我一步」。协程跑完时 `send` 抛 `StopIteration`，`step` 就不再登记。 +- **Loop**：事件循环。`ready` 是就绪队列，放着现在就能执行的回调。`timers` 是按到点时间排序的定时器堆，元组中间的 `id(cb)` 只是为了两个时间相同时，堆不用去比较两个函数（比较函数会报错）。`sleep` 建一个 Future，再定一个到点时调它 `set_result` 的定时器，自己并不睡。`run()` 每一圈做三件事：把到点的定时器挪进就绪队列，就绪队列空了就用 `time.sleep` 睡到最近的定时器，否则取出一个回调执行。 + +#### 逐步走一遍 + +给代码加上打印后，实测（2026-10-25）的执行顺序如下，协程 A 等 0.2 秒，B 等 0.1 秒： + +| 时刻 | 发生了什么 | 之后的就绪队列 | 之后的定时器 | +| --- | --- | --- | --- | +| 0 ms | 建 Task A、Task B，各把自己的 `step` 排进队列 | A.step、B.step | 空 | +| 0 ms | 取出 A.step，调 `send`。协程 A 记下「A 开始」，调 `loop.sleep(0.2)` 拿到 fut_A，并定了一个 200 ms 的定时器。A 执行 `await fut_A`，在 `__await__` 的 `yield` 处停下，`send` 返回 fut_A。A.step 把自己登记到 fut_A 的回调列表里 | B.step | 200 ms 时 fut_A 出结果 | +| 0 ms | 取出 B.step，过程一样：「B 开始」，建 fut_B 和 100 ms 的定时器，停在 `yield`，B.step 登记到 fut_B | 空 | 100 ms 时 fut_B 出结果，200 ms 时 fut_A 出结果 | +| 0 ms | 就绪队列空了，用 `time.sleep` 睡到最近的定时器 | 空 | 同上 | +| 109 ms | 100 ms 的定时器到点，执行 `fut_B.set_result(None)`，把登记着的 B.step 排进队列 | B.step | 200 ms 时 fut_A 出结果 | +| 109 ms | 取出 B.step 调 `send`，协程 B 从 `yield` 下一行接着跑，`__await__` 返回 None，记下「B 结束」。函数跑完，`send` 抛 `StopIteration`，B.step 返回 | 空 | 同上 | +| 109 ms | 就绪队列又空了，睡 91 ms | 空 | 同上 | +| 203 ms | 200 ms 的定时器到点：`fut_A.set_result(None)`，A.step 排进队列，执行后记下「A 结束」 | 空 | 空 | +| 203 ms | 队列和定时器都空了，`run()` 返回 | | | + +时刻是 109 毫秒而不是 100 毫秒，是因为 Windows 上 `time.sleep` 的精度只有 15.6 毫秒左右。 + +#### 串起来的几个要点 + +- **协程暂停是什么：** 执行到 `__await__` 里的 `yield`，局部变量和执行位置都存进协程对象，`send` 返回。CPU 回到 `Task.step`，`step` 返回，再回到 `run()` 的循环里。「把控制权还给事件循环」说的就是这一路返回，没有别的魔法。 +- **谁来唤醒协程：** 没有谁主动去唤醒。Future 出结果时，把当初登记的 `step` 排进队列。事件循环执行到这个 `step`，`step` 调 `send`，协程就接着跑。所以 `await fut` 等于「把『推我一步』登记成 `fut` 的回调，然后一路返回到事件循环」。 +- **为什么两个 `await` 之间不会被打断：** `run()` 一次只取一个回调执行，要等这个回调执行到协程停在下一个 `await`、或者协程结束，才取下一个。全程只有一个线程，别的协程没有机会插进来。 +- **为什么协程里不能用阻塞调用：** 如果 `worker` 里写的是 `time.sleep(0.2)`，`run()` 就卡在执行 A.step 这一个回调里，整整 0.2 秒轮不到 B。`await loop.sleep(0.2)` 则立刻返回到 `run()`，等的这段时间里事件循环照常干别的。 +- **为什么 Future 用 `call_soon`、不直接调回调：** 这样所有代码都由 `run()` 一个一个取出来执行，顺序清楚。如果在 `set_result` 里直接调 `step`，协程 B 就会在定时器回调的里面接着跑，调用一层套一层，顺序也更难推。 + +CPython 3.12 的 `asyncio` 就是这个骨架。`asyncio/futures.py` 里 `Future.__await__` 的核心就是 `yield self`，源码注释写着 "This tells Task to wait for completion"。`asyncio/tasks.py` 里 Task 推进协程的那段，先调 `coro.send(None)`，再对拿到的 Future 调 `add_done_callback(self.__wakeup)`。 + +> [!optional] 别的语言里的协程和 Future。先用两个分类把它们放在一起比。 +> +> 第一个分类是无栈和有栈。Python、JavaScript、C++20、Rust 的协程是无栈协程（stackless coroutine）：暂停时只保存这个函数自己的局部状态，所以只能在 `async` 函数里直接写 `await` 的地方暂停。一个函数里要 `await`，它自己就得是 `async`，调用它的函数也得是 `async`，一路往上传，这叫函数着色（function coloring）。Go 的 goroutine 和 Java 21 的虚拟线程是有栈的：暂停时保存整条调用栈，在任何一层的阻塞调用处都能暂停，所以不需要 `async` 和 `await` 关键字，普通的阻塞写法直接用。 +> +> 第二个分类是协作式和抢占式。Python 和 JavaScript 的协程只在 `await` 处切换，而且都在一个线程上，所以两个 `await` 之间不用加锁。goroutine 和虚拟线程会被分到多个系统线程上同时跑（goroutine 还会被运行时抢占），所以共享数据照样要加锁。 +> +> | 语言 | 写法 | 表示「将来的结果」的对象 | 谁来调度 | +> | --- | --- | --- | --- | +> | Python | `async`/`await` | `asyncio.Future`，以及线程池用的 `concurrent.futures.Future` | 标准库 `asyncio` 的事件循环，一个线程 | +> | JavaScript | `async`/`await`（ES2017） | `Promise`（ES2015），用 `.then()` 接回调，也能 `await` | 运行环境（浏览器或 Node.js）自带的事件循环，JS 代码只在一个线程上跑 | +> | C++ | C++20 的 `co_await`、`co_return` 只是语言机制，标准库里没有现成的任务类型和调度器 | `std::future`（C++11），只能 `get()` 阻塞着等，没有 `.then()` | 用第三方库，比如 Boost.Asio。C++26 的 `std::execution`（senders/receivers）2024 年 6 月投票进了标准草案 | +> | Java | 没有 `async`/`await`，Java 21 的虚拟线程（JEP 444）照常写阻塞代码 | `CompletableFuture`（Java 8），用 `thenApply` 接回调 | JVM 把大量虚拟线程分到少量系统线程上。Java 24 的 JEP 491 让虚拟线程在 `synchronized` 里阻塞时也能让出系统线程 | +> | Go | 没有 `await`，用 goroutine 和阻塞写法 | 一般用 channel 传结果 | Go 运行时把 goroutine 分到多个系统线程上，会抢占 | +> | Rust | `async`/`await` | `Future` trait，惰性：没人去 poll 它就什么都不做 | 标准库只有语言机制，调度靠第三方运行时，比如 tokio | +> +> 表里 C++26 的 `std::execution`（[ACCU 的介绍](https://www.accu.org/journals/overload/32/184/teodorescu/)）和 Java 的 [JEP 491](https://openjdk.org/jeps/491) 是 2026-10-25 查证过的，其余版本号凭记忆，未逐条核实。 ## 速查