Python 并发语法

这一页讲什么

面试里的并发题用 Python 写，要用到的只是标准库 threading 模块里的五样东西：线程、锁、信号量、条件变量、Event，再加一个线程安全的队列 queue.Queue，以及 concurrent.futures 里的线程池 ThreadPoolExecutor。用 asyncio 写时，还要知道协程和 await。这一页每样讲三件事：它是什么、怎么写、拿不到或用错会怎样。题解页（比如 Fire Event：多线程回调）默认你看过这一页，不再重讲语法。

所有代码都在 Python 3.12.10 上跑过（线程池、信号量、GIL、asyncio 四节和 RLock 的补充是 2026-10-25，其余是 2026-10-23），结果写在每段后面。

线程：Thread

线程（thread）

程序里一条独立往下跑的执行流。同一个程序的几个线程共用同一份内存，能读写同样的变量，所以会互相干扰。

在 Python 里建一个 Thread，告诉它要跑哪个函数、带什么参数。调 start() 之后它开始跑，join() 让当前线程停下来等它跑完。

import threading

results = []

def square(n):
    results.append(n * n)

threads = [threading.Thread(target=square, args=(i,)) for i in range(4)]
for t in threads:
    t.start()          # 线程从这里开始跑 square(i)
for t in threads:
    t.join()           # 等它跑完
# sorted(results) == [0, 1, 4, 9]

start() 只能调一次。忘了 join() 的话，主线程可能在子线程跑完之前就往下走了，读到不完整的 results。

锁：Lock

为什么需要

两个线程同时读写同一个变量，结果就取决于它们的指令怎么交错。最简单的例子是计数器：counter += 1 其实是「读出来、加一、写回去」三步，如果线程 A 读完还没写回，线程 B 也读了同一个旧值，两个线程各写回一次，结果只加了一，丢了一次更新。

竞态条件（race condition）

几个线程同时读写同一份数据，至少有一个在写，最后的结果取决于它们的指令恰好怎么交错。上面计数器丢更新就是一个例子。

counter = 0

def add(times):
    global counter
    for _ in range(times):
        v = counter        # 读
        time.sleep(0)      # 让出 CPU，把「读完还没写」的窗口撑大
        counter = v + 1    # 写

threads = [threading.Thread(target=add, args=(200_000,)) for _ in range(4)]
# 四个线程各加 200000 次，期望 800000

实测结果是 200202，丢了 599798 次更新。

去掉那行 time.sleep(0) 再跑，结果恰好是 800000，一次没丢。这并不说明不加锁也安全。原因是 CPython 3.12 只在少数几个固定位置检查要不要切换线程（主要是函数入口和循环往回跳的地方），这段循环里读和写之间没有这样的位置，所以恰好不会在中间切。读和写之间一旦有任何会让出 CPU 的操作（I/O、sleep、等锁、调用 C 写的库），就会丢。面试里不要拿「我跑了没出错」当证据。

怎么写

锁（lock，也叫互斥锁 mutex）

一个对象，只有两个操作。acquire() 是「拿锁」：没人拿着就立刻拿到、继续往下跑。别的线程正拿着，就停在这一行等，直到那个线程放锁。release() 是「放锁」：放掉以后，正在等的线程里有一个会拿到并继续。所以同一时刻最多一个线程拿着锁。

临界区（critical section）

从拿到锁到放锁之间执行的那段代码。拿同一把锁的两个线程，临界区不会同时跑。

counter = 0
lock = threading.Lock()

def add(times):
    global counter
    for _ in range(times):
        with lock:         # 进块时 acquire()，出块时 release()
            counter += 1

同样四个线程各 200000 次，结果 800000。

with lock: 等于进块 acquire()、出块 release()，而且块里抛异常或 return 也会放锁。手写 acquire() / release() 的话，中间抛了异常锁就永远不会被放掉，所以手写时要用 try / finally。面试里一律用 with。

拿不到锁会停在那一行。 实测：线程 A 拿着锁睡 0.3 秒，线程 B 这时调 acquire()，B 停了 0.25 秒才拿到（减去 B 晚起的 0.05 秒，正好是 A 剩下的持锁时间）。

锁只管同样去拿这把锁的代码。 一个线程不拿锁就直接读写变量，锁拦不住它。常有人以为锁是「锁住某个变量」或「锁住整个对象」。其实它只是一个所有线程都约好要先拿的信物，谁不守约，谁就能和别人的临界区同时跑。

同一个线程拿两次：Lock 会死锁，RLock 不会

Lock 不认线程：一个线程拿着锁再调一次 acquire()，它会等「拿锁的人」放锁，而拿锁的人就是它自己，于是永远等下去。这叫不可重入（non-reentrant）。典型场景是一个加了锁的方法里调用另一个也加了同一把锁的方法。

死锁（deadlock）

几个线程互相等对方手里的东西，谁都等不到，全部永远停住。这里是最简单的一种：一个线程等的是它自己手里的锁。

lock = threading.Lock()
lock.acquire()
lock.acquire(timeout=0.1)    # 同一个线程再拿：等满 0.1 秒，返回 False
lock.release()

rlock = threading.RLock()
rlock.acquire()
rlock.acquire(timeout=0.1)   # 同一个线程再拿：立刻返回 True
rlock.release()
rlock.release()              # 拿了两次就要放两次

可重入锁（reentrant lock）

记得自己被哪个线程拿着、拿了几次的锁。同一个线程可以反复拿，每拿一次计数加 1，每放一次减 1，减到 0 才真正放开。Python 里是 threading.RLock。

面试里拿 RLock 解决「锁里调用自己」的问题可以，但要说出代价：它没解决「拿着锁去调别人的代码」带来的其他问题（慢、抛异常），只是不死锁了。

「可重入」只对拿着锁的那个线程成立，这个线程叫锁的主人（owner）。别的线程来 acquire()，和普通的 Lock 一样要等，一直等到主人把锁完全放开：主人拿了几次，就要放几次，计数归零才算放开。别的线程也不能替主人放锁。

rlock = threading.RLock()
rlock.acquire()
rlock.acquire()              # 主线程拿了两次，计数是 2
rlock.release()              # 只放了一次，计数还是 1
# 另一个线程这时调 rlock.acquire(timeout=0.2)：等满 0.2 秒，返回 False
rlock.release()              # 再放一次，计数归零
# 另一个线程再调 acquire：立刻拿到

实测（2026-10-25）：结果和注释一样。主线程拿着锁的时候，另一个线程调 rlock.release()，抛 RuntimeError: cannot release un-acquired lock。

信号量：Semaphore

信号量（semaphore）

一个许可计数。Semaphore(N) 一开始有 N 个许可。acquire() 拿走一个，许可是 0 就停住等。release() 还回一个，正在等的线程里有一个会拿到。它不记得许可被谁拿走了，所以任何线程都可以 release()。

信号量有两种用法。

限制同时做某件事的线程数，比如最多 3 个线程同时下载。用来限流时写 BoundedSemaphore，原因见下面「用错会怎样」。

slots = threading.BoundedSemaphore(3)

def download(url):
    with slots:                  # 进块时 acquire，出块时 release
        fetch(url)               # 同一时刻最多 3 个线程在这里

实测：8 个线程，每个在里面待 50 毫秒，同一时刻最多 3 个在里面，总共 152 毫秒，正好是分 3 批进去的时间。

一个线程通知另一个线程。初始值设成 0，等的一方 acquire()，通知的一方做完事后 release()。LeetCode 1114（Print in Order）就是这类题：

first_done = threading.Semaphore(0)

def first():
    print("first")
    first_done.release()         # 通知：first 做完了

def second():
    first_done.acquire()         # first 还没做完，就停在这里等
    print("second")

实测：先启动调 second 的线程，50 毫秒后它还什么都没打印。再启动调 first 的线程，输出顺序是 first、second。

用错会怎样：

• release() 多了不报错。 普通的 Semaphore 不管初始值是多少，多放一次许可就多一个，等于多放进一个线程，程序不报任何错。实测 Semaphore(1) 多放一次以后，能连拿两次。BoundedSemaphore 会检查，放的次数超过初始值时抛 ValueError: Semaphore released too many times（实测）。
• 同一个线程拿两次会卡住。 信号量不认线程，Semaphore(1) 在同一个线程里第二次 acquire() 会一直等（实测等满 0.2 秒，返回 False）。许可不止一个时，同一个线程能连拿几次，就看还剩几个许可：实测 Semaphore(2) 连拿两次都成功，第三次等满 0.1 秒返回 False。

和 Lock、RLock 的区别

表里容易看漏的一格：Python 的 Lock 也不记主人，文档写明任何线程都能调它的 release()，实测也是这样。所以在 Python 里，Semaphore(1) 用起来和 Lock 很像，都是一次只放一个线程进去。三者里只有 RLock 检查「放锁的是不是拿锁的那个线程」。

面试里怎么选：

• 保护共享数据，用 Lock。
• 锁里会调到同样要拿这把锁的代码，用 RLock。上一节说过它只解决「自己卡死自己」，所以先想想能不能把代码理顺。
• 限制同时 N 个，用 BoundedSemaphore(N)。
• 线程之间按顺序通知，用初始为 0 的 Semaphore。

An RLock is a mutex with an owner. The same thread can acquire it again, and other threads can take it only after the owner has released it as many times as it acquired it. A semaphore is a counter of permits with no owner, so any thread can release it. I'd use it to cap concurrency at N, or to signal between threads.

条件变量：Condition

为什么需要

锁解决的是「别同时改」。另一类需求是「等某件事发生再继续」：消费者要等队列里有东西，Trigger 题里 wait() 要等有人 fire()。最笨的办法是忙等（busy wait）：while not flag: pass 一直空转查标志，白白占满一个 CPU 核。条件变量能让线程睡下去、不占 CPU，等别人叫醒。

条件变量（condition variable）

一个和锁绑在一起用的等待工具。线程在它上面 wait()，放锁并睡下。别的线程改完条件以后 notify() 或 notify_all()，把等着的线程叫醒。它只负责「睡」和「叫醒」，条件成没成立要另用一个变量记下来。

怎么写

Condition 内部自带一把锁，with cond: 拿的就是这把锁。它提供：

• wait()：必须在拿着锁的时候调用。它在一步之内放锁并让线程睡下。被叫醒以后，要重新拿到锁才返回。「放锁」和「睡下」必须是一步，否则线程刚放锁、还没睡着，别人就改了条件并发了通知，这个通知没人收到，线程随后睡下去就再也醒不过来（这叫丢失唤醒，见下面的名词解释）。
• notify()：叫醒一个正在 wait() 的线程，notify_all() 叫醒全部。都必须在拿着锁的时候调用，不拿会直接抛 RuntimeError: cannot notify on un-acquired lock。它们不留任何记录，只叫醒发通知那一刻正在等的线程：没人在等时这次通知就落空了，之后才来 wait() 的线程照样睡下，等的是下一次通知（实测先 notify_all()、再 wait(timeout=0.3)，睡满 0.3 秒超时返回 False）。所以「发生过什么」要另用一个变量记下来，比如下面例子里的 items，等的一方先查它，再决定要不要睡。前面的 Semaphore 和后面的 Event 会记住之前的 release()、set()，之后再来的线程立刻返回。

丢失唤醒（lost wakeup）

通知发出的时候，要等的线程还没开始等，这次通知就落空了。那个线程随后开始等，可再也没有人通知它，于是永远睡下去。

cond = threading.Condition()      # 内部自带一把锁
items = []

def consumer():
    with cond:
        while not items:          # 用 while：醒来后重新检查
            cond.wait()           # 放锁、睡下；被叫醒后重新拿到锁才返回
        item = items.pop()

# 生产者：
with cond:
    items.append("job")           # 先在锁里改条件
    cond.notify()                 # 再叫醒

实测：消费者先进入 wait()，生产者放入一个元素并 notify() 后，消费者拿到了它。

虚假唤醒（spurious wakeup）

等待的线程在没有任何人通知的情况下，从 wait() 里返回了。POSIX 线程库和 C++ 标准都明确允许这种情况，所以从 wait() 返回只说明「该醒来再看一眼了」，不说明条件已经成立。

wait() 外面用 while 不用 if，两个原因。第一，可能是虚假唤醒，醒来不代表条件成立。第二，从被叫醒到重新拿到锁之间有一段空档，别的线程可能在这段时间里把条件又改回去了（比如另一个消费者先一步把元素拿走了）。所以醒来后一律重新检查，不成立就接着睡。

先改条件、再通知，两步都在锁里。 顺序反过来（先通知、再在锁外改条件），被叫醒的线程重新拿到锁时查到的还是旧条件，又睡回去，之后再也没人叫它。Fire Event：多线程回调 思路第 8 节有这个错误的逐步表。

它自己是怎么用锁实现的

threading.Condition 是用 Python 写的，源码在标准库的 threading.py 里，只用到了锁。思路是每个等待的线程配一把自己的小锁，当作一扇门：wait() 先把门锁上，再撞一次门把自己卡住，notify() 替它把门打开。下面是照着源码写的简化版，去掉了超时、notify_all() 和 RLock 的处理：

import threading
from collections import deque

class MyCondition:
    def __init__(self):
        self._lock = threading.Lock()          # 外层锁，保护条件
        self._waiters = deque()                # 每个等待的线程一把小锁

    def __enter__(self):
        self._lock.acquire()
        return self

    def __exit__(self, *exc):
        self._lock.release()

    def wait(self):                            # 调用时必须拿着外层锁
        waiter = threading.Lock()
        waiter.acquire()                       # 先把小锁锁上
        self._waiters.append(waiter)           # 还拿着外层锁时登记
        self._lock.release()                   # 放外层锁
        waiter.acquire()                       # 再拿一次小锁：卡住，直到 notify 替它放掉
        self._lock.acquire()                   # 醒来后重新拿外层锁才返回

    def notify(self):                          # 调用时必须拿着外层锁
        if self._waiters:
            self._waiters.popleft().release()  # Lock 不记主人，别的线程可以放

它用到了这一页讲过的两个性质。一是普通 Lock 同一个线程拿第二次会卡住（「同一个线程拿两次」那一节），wait() 靠这个让线程睡下。二是 Lock 不记主人（「和 Lock、RLock 的区别」那张表），所以 notify() 能替等待的线程把它的小锁放掉。

上面「怎么写」里说「放锁」和「睡下」必须是一步，这段代码里两者却是分开的两行。它不丢唤醒，靠的是登记的时机：等待的线程还拿着外层锁的时候，就把自己的小锁放进了 _waiters。通知方要先拿到外层锁才能 notify()，所以它一定能看到这把小锁。就算 notify() 正好发生在「放外层锁」和「再拿小锁」之间，小锁也已经被放开了，等待的线程第二次拿时立刻拿到，不会睡死。从外面看，这两行和「一步」的效果一样。这和 Fire Event：多线程回调 里 register 的道理相同：检查和登记必须在同一次持锁里完成。

实测（2026-10-25）：用它写和上面一样的生产者、消费者，在「放外层锁」和「再拿小锁」之间停 50 毫秒，让生产者正好在这个空档里 notify()，10 次消费者都拿到了元素。不停的话也是 10 次都拿到。

标准库的 Condition 骨架和这个一样，多处理了几件事：外层锁默认是 RLock，wait() 放锁时要连同重入的次数一起放掉，醒来再恢复。wait() 可以带超时。notify_all() 把所有小锁都放开。

Event：现成的一次性开关

threading.Event 就是一个封装好的「标志 + 条件变量」，有四个方法：

• wait() 阻塞到有人 set()，已经 set() 过就立刻返回。
• set() 叫醒所有等待者。
• is_set() 查状态。
• clear() 复位。

ready = threading.Event()

def waiter():
    ready.wait()                  # 阻塞到有人 set()；已经 set 过就立刻返回
    ...

# 三个线程先 wait()，都停住；主线程 ready.set() 后三个都醒来；之后再 wait() 立刻返回

实测：三个等待线程在 set() 前都停住（is_set() 为 False），set() 后全部醒来，之后再 wait() 立刻返回。

它正是 Trigger 题的标准答案。面试里面试官要的是你自己用 Condition 实现它，所以不能直接拿来用，但你应该知道它存在、知道自己实现的就是它，这能说明你熟悉标准库。

线程安全的队列：queue.Queue

生产者 / 消费者之间传东西，用 queue.Queue，它内部已经加好了锁和条件变量：put() 放入，get() 取出，队列空的时候 get() 停住等。常用的收尾约定是放一个 None 表示「下班」。

import queue

q = queue.Queue()

def worker():
    while True:
        item = q.get()            # 空的时候停住等
        if item is None:          # 约定 None 表示下班
            return
        handle(item)

# 主线程：q.put(1)、q.put(2)、……、最后 q.put(None)

实测：放入 0 到 4 再放 None，工作线程按顺序取到 0 到 4 后退出。

面试里如果题目本身就是「实现一个线程安全的队列」或「实现生产者消费者」，不能用它。题目是别的、队列只是工具时，用它比自己写省事也不容易错。

线程池：ThreadPoolExecutor

线程池（thread pool）

一组事先建好、反复使用的工作线程（worker thread）。你把任务提交给它，它从内部的队列里按提交顺序取出任务，交给空闲的工作线程执行。

和每个任务新建一个 Thread 相比，线程只建一次、反复使用，同时执行的任务数也有上限。标准库的线程池在 concurrent.futures 模块里，叫 ThreadPoolExecutor。

Future 对象

代表一个已经提交、可能还没执行完的任务。任务执行完以后，返回值或者抛出的异常都存在它里面。用 result() 取返回值，任务还没执行完就停住等。用 exception() 取异常对象。

submit(f, *args) 把一个任务交给线程池，马上返回一个 Future。wait(futures) 等一组任务全部执行完。

import time
from concurrent.futures import ThreadPoolExecutor, wait

def slow_square(n):
    time.sleep(0.1)
    return n * n

with ThreadPoolExecutor(max_workers=4) as pool:      # 出 with 时自动 shutdown(wait=True)
    futures = [pool.submit(slow_square, i) for i in range(4)]   # submit 马上返回 Future
    wait(futures)                                    # 等这 4 个任务全部执行完
    results = [f.result() for f in futures]          # result() 取返回值
# results == [0, 1, 4, 9]

实测：results 是 [0, 1, 4, 9]，总共 101 毫秒。4 个任务同时在 4 个工作线程上 sleep，所以总时间约等于一个任务的时间，串行要 400 毫秒。

用错会怎样：

• 任务抛的异常不会自己出现。 异常存在 Future 里，submit 那一行不会报错，屏幕上也不打印。要调 f.result()（重新抛出这个异常）或 f.exception()（把异常对象返回给你）才看得到。实测提交一个抛 ValueError 的函数，什么都没打印，f.exception() 返回 ValueError('boom')。
• 关了以后不能再提交。 shutdown() 之后再 submit，抛 RuntimeError: cannot schedule new futures after shutdown（实测）。用 with 写，出块时会自动关。

max_workers=1 的线程池只有一个工作线程，任务一个执行完才轮到下一个，执行顺序就是提交顺序。实测提交 5 个任务，前面的任务执行时间更长（50 到 10 毫秒），执行顺序仍然是 0、1、2、3、4。需要「所有任务在同一个线程上、按顺序执行」时，就用它，Fire Event：多线程回调 的 Follow-up 第三条就是这么用的。

GIL：多线程什么时候有用

全局解释器锁（GIL，global interpreter lock）

标准的 Python 解释器 CPython 里的一把锁：同一时刻只有一个线程能执行 Python 字节码。

GIL 带来两个后果。一是 Python 线程做纯计算不会并行，只是轮流。二是一条赋值不会写到一半被打断，别的线程也马上能看到。但它不保证两条语句之间不被切开，上面计数器的例子就是证明。所以 GIL 不能代替锁，面试里也别说「Python 有 GIL 所以不用加锁」。

GIL 只限制「执行 Python 字节码」这一件事。线程在等 I/O 的时候不执行字节码，C 代码放掉 GIL 以后在里面算，也不受它限制。所以多线程在三种情况下有用，只有纯 Python 计算帮不上忙。实测（2026-10-25，Python 3.12.10，16 核的 Windows 机器，每行 4 个任务）：

多线程有用的三种情况：

• 等 I/O。 网络请求、读写文件、查数据库都算。线程在等的时候会放掉 GIL，别的线程接着跑，4 个 200 毫秒的等待叠在一起，总共只要 201 毫秒。同时调好几个 API、Web 服务器同时处理多个请求、爬虫，都是这一类。Python 多线程主要用在这里。
• 调用会放掉 GIL 的 C 代码。 hashlib、zlib、NumPy 的大部分运算是 C 写的，它们在 C 里处理大块数据时不碰 Python 对象，会先放掉 GIL 再算，几个线程就能同时用几个核。实测 hashlib 用 4 个线程快了 3 倍多。zlib 和 NumPy 没有实测。
• 保持响应。 一个线程在做耗时的事，别的线程照样能干活，比如 GUI 程序把耗时任务放到后台线程，界面还能响应点击。总时间不变，但程序不卡。正在跑的线程每隔一小段时间（默认 5 毫秒，用 sys.getswitchinterval() 能查到）就会被要求交出 GIL，所以一个线程算个不停，也不会让别的线程一直拿不到 GIL。

纯 Python 计算一直在执行字节码，4 个线程只能轮流跑，740 毫秒和串行的 759 毫秒差不多。这种情况用多进程：每个进程有自己的解释器和自己的 GIL，能同时用多个核。写法和线程池一样，把 ThreadPoolExecutor 换成 concurrent.futures.ProcessPoolExecutor：

from concurrent.futures import ProcessPoolExecutor

def cpu_task(n):                 # 任务函数要定义在模块顶层，子进程按名字找到它
    ...

if __name__ == "__main__":       # Windows 和 macOS 上必须有这一行
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(cpu_task, range(4)))

Windows 和 macOS 上，新进程会重新导入主程序文件。没有 if __name__ == "__main__": 这一行，子进程一导入就又去建进程池，Python 会直接报错。实测去掉这一行，子进程抛 RuntimeError，提示当前进程还没完成启动就试图启动新进程。

多进程的代价有三个：

• 起进程慢。 Windows 上每个新进程都要重新启动解释器、导入模块。表里的 198 毫秒是进程池启动好以后测的，刚建好的进程池头两轮分别用了 501 毫秒和 356 毫秒。
• 参数和结果要序列化。 它们先用 pickle 转成字节，再在进程之间传，数据大的时候这一步也要时间。
• 进程之间不共享内存。 一个进程改了自己的变量，别的进程看不到。

GIL 也可以去掉。Python 3.13 起有一个可选的自由线程构建（free-threaded build，可执行文件叫 python3.13t），没有 GIL，纯 Python 计算也能同时用上多个核。3.14 起它是官方正式支持的版本（PEP 779，2025-06 被接受），但仍然要单独安装，默认的 python 还带着 GIL。PEP 779 给它定的要求是单线程最多慢 10%，内存最多多用 15%。这个构建没有实测。

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 怎么写」，这一节讲这两件事。

协程（coroutine）

用 async def 定义的函数，调用它不会马上执行，要交给事件循环去跑。它可以在 await 的地方暂停，把控制权交出去，等要等的东西好了，再从暂停的地方接着往下执行。

事件循环（event loop）

asyncio 里负责调度协程的那个循环。它在一个线程上反复做一件事：挑一个可以继续执行的协程，让它跑到下一个 await，再挑下一个。asyncio.run(main()) 会建一个事件循环，跑完 main 再关掉。

怎么写

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）：操作系统和解释器可以在任意两条字节码之间把一个线程停下，换另一个线程跑，你控制不了在哪里切。所以要用锁把几步包成临界区。协程是另一种调度方式：

协作式调度（cooperative scheduling）

只有正在执行的那一方主动交出控制权，才会切换到别的任务。在 asyncio 里，主动交出的地方就是 await。两个 await 之间的代码执行时，别的协程插不进来。

所以在 asyncio 里，一段不含 await 的代码天然就像一个临界区，不用加锁。Fire Event：多线程回调 的 Follow-up 第八条里，asyncio 版本不加锁也是对的，原因就在这里。

用错会怎样

临界区跨过 await，照样有竞态

在「检查标志」和「入队」之间等了一次 I/O，别的协程就能在这个 await 处插进来 fire，和线程里的 check-then-act 是同一种错：

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 都在它里面改共享状态：

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）会让整个线程停下，事件循环也就跑不了别的协程：

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（实测）。

协程和线程要打交道，有两个标准写法：

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)。

这几个写法背后是同一种思路：

线程封闭（thread confinement）

共享数据只让一个线程读写，别的线程不碰它，要改就把操作交给那个线程去做。数据只有一个线程碰，就不需要锁。

asyncio 用的就是线程封闭：状态只属于事件循环那个线程，别的线程通过 call_soon_threadsafe 或 run_coroutine_threadsafe 把操作交过去。Fire Event：多线程回调 Follow-up 第三条的做法二（所有回调交给一个工作线程）也是这个思路。

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 和事件循环。协程能「停下、过一会儿再接着跑」，靠的是 Python 的生成器。所以这一节先讲生成器，再看协程怎么停下、怎么接着跑，最后从零写一个 40 行的最小事件循环，逐步走一遍。看懂这一节，await 底下发生了什么就清楚了。

先认识生成器

生成器（generator）

调用一个函数体里写了 yield 的函数，会得到一个生成器对象，这时函数体一行都还没执行。这个对象能跑一段、停下、交出一个值，下次被要求时，再从停下的地方接着跑。普通函数一旦返回，局部变量就没了。生成器停下时，局部变量和跑到了哪一行，都留在生成器对象里。

yield 就是生成器里「停在这里，并把后面这个值交给外面」的意思。外面用 next(g) 要下一个值，g 就从上次停下的地方接着跑，跑到下一个 yield 停下：

def count_to_two():
    print("  生成器里：开始")
    yield 1
    print("  生成器里：从第一个 yield 后面接着跑")
    yield 2
    print("  生成器里：跑到函数末尾")

g = count_to_two()                 # 只建了生成器对象，函数体一行都没执行
print("外面：建好了", type(g).__name__)
print("外面：拿到", next(g))       # 跑到第一个 yield 停下，把 1 交出来
print("外面：拿到", next(g))       # 从停下的地方接着跑，到第二个 yield 停下
try:
    next(g)                        # 接着跑到函数末尾，没有值可交了
except StopIteration:
    print("外面：收到 StopIteration，生成器跑完了")

实测输出：

外面：建好了 generator
  生成器里：开始
外面：拿到 1
  生成器里：从第一个 yield 后面接着跑
外面：拿到 2
  生成器里：跑到函数末尾
外面：收到 StopIteration，生成器跑完了

• count_to_two() 只建了生成器对象，函数体一行都没执行，所以「开始」排在「建好了」后面。
• 第一次 next(g) 从函数开头跑，打印「开始」，到 yield 1 停下，把 1 交给外面。
• 第二次 next(g) 从 yield 1 的下一行接着跑，打印一行，到 yield 2 停下，交出 2。
• 第三次 next(g) 接着跑到函数末尾，已经没有值可交了，于是抛出 StopIteration。

StopIteration

Python 内置的一个异常，意思是「没有下一个值了」。生成器的函数体跑到末尾或者执行了 return，再对它调 next() 或 send()，就会抛出它。return 后面的值放在这个异常的 value 属性里。它是一个正常的结束信号，不是出错。

for 循环背后就是这件事：反复调 next，收到 StopIteration 就结束循环。所以平时用 for 遍历生成器，从来看不到这个异常：

for x in count_to_two():           # for 循环就是反复调 next，收到 StopIteration 就结束
    print("for 拿到", x)

实测打印「for 拿到 1」「for 拿到 2」，中间夹着生成器里的那几行，然后循环自己结束了。

外面除了用 next 要值，还能用 send 往生成器里传一个值进去。yield 本身也是一个表达式：它交出一个值，停下，等外面下次 send(v) 时，v 就成了这个 yield 表达式的值：

def echo():
    received = yield "第一次交出的值"
    print("  生成器里：yield 表达式的值是", received)
    return "函数的返回值"

g = echo()
print("外面：拿到", g.send(None))   # 第一次只能 send(None)，效果和 next(g) 一样
try:
    g.send("你好")                  # "你好" 成了 yield 表达式的值，然后函数跑到 return
except StopIteration as e:
    print("外面：StopIteration，返回值在 e.value 里：", e.value)

实测：第一次拿到「第一次交出的值」，生成器里打印「yield 表达式的值是 你好」，最后外面收到 StopIteration，e.value 是「函数的返回值」。第一次只能 send(None)，因为生成器还没跑到任何 yield，没有地方接收传进来的值，这时 send(None) 和 next(g) 效果一样。

协程的返回值也是这样交出来的：

async def add_one():
    return 41 + 1

coro = add_one()
try:
    coro.send(None)                 # 没有 await，一次就跑到 return
except StopIteration as e:
    print("协程的返回值也在 StopIteration 里：", e.value)

实测打印「协程的返回值也在 StopIteration 里： 42」。asyncio 的 Task 就是靠接住这个异常，拿到协程的结果。

协程怎么停下、怎么接着跑

协程对象也有 send()，用法和生成器一样。区别是协程的函数体里不写 yield，它在 await 处停下。await 后面能写什么，有一个专门的说法：

可等待对象（awaitable）

能写在 await 后面的对象：协程、asyncio 的 Task 和 Future，或者任何定义了 __await__ 方法的对象。__await__ 要返回一个迭代器，通常就是一个生成器。

挂起点（suspension point）

协程执行到这里会停下、把控制权交给外面的那个位置。在 Python 里，await x 会调用 x.__await__()，协程就停在这个生成器里的 yield 上。yield 出来的值穿过协程，成为外面那次 send() 的返回值。外面下次再 send()，协程就从这个 yield 的下一行接着跑。

下面的 Box 就是一个最小的可等待对象（实测 isinstance(Box(), collections.abc.Awaitable) 是 True，Box().__await__() 是一个生成器）：

class Box:                               # 一个最小的「可以被 await 的东西」
    def __init__(self):
        self.done, self.value = False, None
    def __await__(self):
        if not self.done:
            yield self                   # 挂起点：协程停在这里，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. 外面：建好了协程对象")
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. 外面：建好了协程对象
1. 协程里：开始，跑到 await 之前
2. 外面：send 返回了 box
3. 协程里：恢复了，x = 42
4. 外面：第二次 send 抛了 StopIteration，协程跑完了

1 和 3 是协程 demo 里面打印的，0、2、4 是协程外面那几行打印的。按执行的先后，控制权在外面和协程里面之间来回切换：

协程自己没法恢复自己，谁调 send，谁就决定它什么时候接着跑。事件循环做的，就是表里第 2 步和第 4 步「调 send」的那个角色。

一个最小的事件循环

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 的那个写法。这两个方法和 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 秒：

时刻是 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)。

别的语言里的协程和 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 的介绍）和 Java 的 JEP 491 是 2026-10-25 查证过的，其余版本号凭记忆，未逐条核实。

速查