更新页面 Python 并发语法
页面 kb/python-concurrency · 修改（rev 11） · · 2026-10-25 22:54
body
@@ -1,7 +1,7 @@ ## 这一页讲什么 -面试里的并发题用 Python 写，要用到的只是标准库 `threading` 模块里的五样东西：线程、锁、信号量、条件变量、Event，再加一个线程安全的队列 `queue.Queue`，以及 `concurrent.futures` 里的线程池 `ThreadPoolExecutor`。这一页每样讲三件事：它是什么、怎么写、拿不到或用错会怎样。题解页（比如 [[q:everpure-fire-event]]）默认你看过这一页，不再重讲语法。 - -所有代码都在 Python 3.12.10 上跑过（线程池、信号量、GIL 三节和 RLock 的补充是 2026-10-25，其余是 2026-10-23），结果写在每段后面。 +面试里的并发题用 Python 写，要用到的只是标准库 `threading` 模块里的五样东西：线程、锁、信号量、条件变量、Event，再加一个线程安全的队列 `queue.Queue`，以及 `concurrent.futures` 里的线程池 `ThreadPoolExecutor`。用 `asyncio` 写时，还要知道协程和 `await`。这一页每样讲三件事：它是什么、怎么写、拿不到或用错会怎样。题解页（比如 [[q:everpure-fire-event]]）默认你看过这一页，不再重讲语法。 + +所有代码都在 Python 3.12.10 上跑过（线程池、信号量、GIL、asyncio 四节和 RLock 的补充是 2026-10-25，其余是 2026-10-23），结果写在每段后面。 ## 线程：`Thread` @@ -395,4 +395,153 @@ > [!say] The GIL only prevents threads from running Python bytecode at the same time. Threads still help with I/O-bound work, because a thread releases the GIL while it waits, and with C extensions like hashlib or NumPy that release the GIL during heavy computation. For CPU-bound pure Python code, I'd use a process pool instead. + +## 协程和 await：`asyncio` + +前面几节讲的都是线程。Python 还有另一种让很多事情同时进行的办法：标准库的 `asyncio`。面试里常被问「`await` 和锁有什么区别」「这题用 `asyncio` 怎么写」，这一节讲这两件事。 + +> [!term] 协程（coroutine） +> 用 `async def` 定义的函数，调用它不会马上执行，要交给事件循环去跑。它可以在 `await` 的地方暂停，把控制权交出去，等要等的东西好了，再从暂停的地方接着往下执行。 + +> [!term] 事件循环（event loop） +> `asyncio` 里负责调度协程的那个循环。它在一个线程上反复做一件事：挑一个可以继续执行的协程，让它跑到下一个 `await`，再挑下一个。`asyncio.run(main())` 会建一个事件循环，跑完 `main` 再关掉。 + +### 怎么写 + +```python +import asyncio + +async def fetch(name, seconds): + print(name, "开始") + await asyncio.sleep(seconds) # 挂起这个协程，事件循环去跑别的协程 + print(name, "结束") + return name + +async def main(): + results = await asyncio.gather(fetch("A", 0.2), fetch("B", 0.1)) + print(results) + +asyncio.run(main()) # 建一个事件循环，跑 main，跑完关掉 +``` + +实测：输出是 A 开始、B 开始、B 结束、A 结束，最后是 `['A', 'B']`，总共 199 毫秒。A 在 `await` 处暂停以后，事件循环转去跑 B，所以两个等待叠在一起，总时间接近较长的那个 0.2 秒。`asyncio.gather` 同时跑几个协程，等它们都结束，按传入的顺序返回结果。 + +### `await` 和锁的区别 + +线程是抢占式调度（preemptive scheduling）：操作系统和解释器可以在任意两条字节码之间把一个线程停下，换另一个线程跑，你控制不了在哪里切。所以要用锁把几步包成临界区。协程是另一种调度方式： + +> [!term] 协作式调度（cooperative scheduling） +> 只有正在执行的那一方主动交出控制权，才会切换到别的任务。在 `asyncio` 里，主动交出的地方就是 `await`。两个 `await` 之间的代码执行时，别的协程插不进来。 + +所以在 `asyncio` 里，一段不含 `await` 的代码天然就像一个临界区，不用加锁。[[q:everpure-fire-event]] 的 Follow-up 第八条里，`asyncio` 版本不加锁也是对的，原因就在这里。 + +| | 线程加锁 | 协程加 `await` | +| --- | --- | --- | +| 什么时候切换 | 任意两条字节码之间都可能 | 只在 `await` 处 | +| 怎么让几步不被打断 | 用锁把它们包成临界区 | 这几步之间不写 `await` | +| 等待时 | 整个线程睡下 | 只有这个协程挂起，线程去跑别的协程 | +| 等错了地方会怎样 | 拿着锁等别人，死锁 | 在协程里调阻塞函数，整个事件循环卡住 | +| 适合什么 | 要调用会阻塞的库，或者代码本来就是多线程 | 大量同时进行的 I/O，比如几千个网络连接 | + +### 用错会怎样 + +#### 临界区跨过 `await`，照样有竞态 + +在「检查标志」和「入队」之间等了一次 I/O，别的协程就能在这个 `await` 处插进来 fire，和线程里的 check-then-act 是同一种错： + +```python +async def register(self, cb): # 错误写法：检查和入队之间有 await + if not self._fired: + await asyncio.sleep(0) # 代表中间要等一次 I/O + self._queue.append(cb) + else: + cb() +``` + +修法是用 `asyncio.Lock`，`register` 和 `fire` 都在它里面改共享状态： + +```python +class AsyncEventCallbacks: + def __init__(self): + self._fired = False + self._queue = deque() + self._lock = asyncio.Lock() + + async def register(self, cb): + async with self._lock: # 临界区里有 await，要用 asyncio.Lock + if not self._fired: + await asyncio.sleep(0) # 代表中间要等一次 I/O + self._queue.append(cb) + return + cb() + + async def fire(self): + async with self._lock: + self._fired = True + pending, self._queue = self._queue, deque() + for cb in pending: + cb() +``` + +实测（2026-10-25）：错误写法和一个 fire 的协程同时跑，10 次回调全丢，都留在队列里。用 `asyncio.Lock` 的写法 10 次全对。`asyncio` 还有 `asyncio.Event`、`asyncio.Condition`，用法和 `threading` 里同名的东西很像。它们只在同一个事件循环里的协程之间有效，不能拿来同步线程。 + +#### 在协程里调阻塞函数，整个事件循环都会停 + +`time.sleep`、`threading.Lock.acquire`、`queue.Queue.get` 这类阻塞调用（blocking call）会让整个线程停下，事件循环也就跑不了别的协程： + +```python +async def blocking(): + time.sleep(0.1) # 阻塞调用：整个线程停 0.1 秒，事件循环也停了 + +async def cooperative(): + await asyncio.sleep(0.1) # 只挂起这个协程，事件循环照常跑别的协程 +``` + +实测（2026-10-25）：让一个每 20 毫秒打一次点的协程和它们同时跑。和 `blocking` 一起跑时，第 2 次打点被推迟到了第 100 毫秒。和 `cooperative` 一起跑时，打点间隔照常（Windows 上 `sleep` 的精度约 15.6 毫秒，所以实际间隔是 31 毫秒左右）。所以协程里要等，就用 `await`，不用阻塞调用。 + +#### 普通函数里不能 `await`，线程锁也不能被 `await` + +在不是 `async def` 的函数里写 `await`，是语法错误：`SyntaxError: 'await' outside async function`。`await threading.Lock()` 会抛 `TypeError: object _thread.lock can't be used in 'await' expression`（都实测过）。所以多线程的题目用不上 `await`，题目一旦说「多个线程同时调用」，就得用 `threading` 的锁。 + +### 协程和线程的关系 + +一个事件循环只在一个线程上跑，它管的协程也都在这个线程上，在 `await` 前后也不会换线程。程序可以开好几个事件循环，每个占一个线程。实测两个线程各自 `asyncio.run` 一个循环，循环 1 的协程始终在线程 1 上，循环 2 的始终在线程 2 上。协程、`Future`、`asyncio.Lock` 都绑定在创建它们的那个循环上，在别的循环里 `await` 一个 `Future` 会报 `RuntimeError: ... attached to a different loop`（实测）。 + +协程和线程要打交道，有两个标准写法： + +```python +import asyncio, threading, time + +def blocking_io(): + time.sleep(0.1) # 比如一个只有同步接口的库 + return threading.current_thread().name + +async def main(): + # 协程要调用阻塞函数：交给线程池，自己 await 结果 + print(await asyncio.to_thread(blocking_io)) + +asyncio.run(main()) + +# 别的线程要让某个事件循环执行协程：交给那个循环，拿回一个 Future +loop = asyncio.new_event_loop() +threading.Thread(target=loop.run_forever, daemon=True).start() + +async def where(): + return threading.current_thread().name + +future = asyncio.run_coroutine_threadsafe(where(), loop) +print(future.result(timeout=1)) # 在主线程里等结果 +loop.call_soon_threadsafe(loop.stop) +``` + +实测：`to_thread` 里的函数跑在线程池的线程 `asyncio_0` 上，协程在原来的线程上等结果，等的时候事件循环照常跑别的协程。从主线程用 `run_coroutine_threadsafe` 交过去的协程，跑在那个循环所在的线程上。要交过去的如果是普通函数，用 `loop.call_soon_threadsafe(f, *args)`。 + +这几个写法背后是同一种思路： + +> [!term] 线程封闭（thread confinement） +> 共享数据只让一个线程读写，别的线程不碰它，要改就把操作交给那个线程去做。数据只有一个线程碰，就不需要锁。 + +`asyncio` 用的就是线程封闭：状态只属于事件循环那个线程，别的线程通过 `call_soon_threadsafe` 或 `run_coroutine_threadsafe` 把操作交过去。[[q:everpure-fire-event]] Follow-up 第三条的做法二（所有回调交给一个工作线程）也是这个思路。 + +> [!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. ## 速查 @@ -410,2 +559,6 @@ | 把任务交给一组线程执行 | `ThreadPoolExecutor` | `with ThreadPoolExecutor(max_workers=4) as pool:` 里 `f = pool.submit(fn, x)`，再 `f.result()` | | 纯 Python 计算要用多个核 | `ProcessPoolExecutor` | 写法同上，放在 `if __name__ == "__main__":` 里面 | +| 大量同时进行的 I/O，不想开很多线程 | `asyncio` | 函数写成 `async def`，要等的地方 `await`，入口 `asyncio.run(main())` | +| 协程之间的临界区里有 `await` | `asyncio.Lock` | `async with lock: ...` | +| 协程里要调阻塞函数 | `asyncio.to_thread` | `await asyncio.to_thread(f, x)` | +| 别的线程要让事件循环执行协程 | `run_coroutine_threadsafe` | `asyncio.run_coroutine_threadsafe(coro, loop).result()` |