S10-X14 · 并发原语：锁、队列与信号

Python 的并发问题有一半不是并发本身，而是选错了执行模型：CPU 密集任务丢进线程池，跑得比单线程还慢；协程里写了一句 time.sleep，整个事件循环停在原地。

这篇按"选模型 → 用原语 → 处理超时/取消 → 死锁"的顺序过一遍，每个结论都落到具体 API。

1. GIL 决定了什么

CPython 的全局解释器锁保证同一时刻只有一个线程执行字节码。三条直接后果：

任务类型	该用什么	理由
CPU 密集（计算、序列化、图像处理）	多进程	多线程拿不到并行，只多出切换开销
IO 密集（网络、磁盘）	线程或协程	等待期间 GIL 释放，可重叠

GIL 不是"Python 慢"的主因，但它划定了线程池的边界：线程数超过核数不提升 CPU 吞吐，只增加切换开销。

版本演进要按版本核对：

版本	变化	影响
3.11	专用化自适应解释器	单线程更快，GIL 语义未变
3.12	支持"每解释器独立 GIL"（PEP 684，C API 层）	子解释器可并行，`threading` 不自动利用
3.13	提供实验性 free-threaded 构建（PEP 703）	无 GIL 版本，非默认，生态兼容性需验证

结论不变：默认构建上，多线程不做并行计算。要用 3.13+ 的 free-threaded 模式，需同时满足"该构建、依赖已适配、实测有收益"。

2. threading 的五种同步原语

原语	语义	适用
`Lock`	二值互斥，不可重入	保护临界区、一次性令牌
`RLock`	可重入，记录持有线程与次数	递归调用、同类方法互相调用
`Semaphore`	计数许可，可跨线程释放	限流、限制并发连接数
`Event`	布尔标志 + 等待	启动信号、一次性通知
`Condition`	锁 + 等待队列	生产者消费者、复杂等待条件

import threading, time

lock = threading.Lock()
counter = 0

def bump():
    global counter
    with lock:              # with 就是 acquire/release 的糖
        counter += 1

ts = [threading.Thread(target=bump) for _ in range(100)]
[t.start() for t in ts]
[t.join() for t in ts]
print(counter)

Lock 不可重入：同一线程第二次 acquire() 会永久阻塞（非超时模式下）。表现是"程序卡死但没有 CPU 占用"。同一线程里嵌套获取同一把锁，必须用 RLock。

Semaphore 允许非持有者 release()，这让它适合当计数器，也容易被误用成"释放次数不对"。

3. queue：三兄弟与 join 的配对

类	顺序	适用
`Queue`	FIFO	通用任务队列
`LifoQueue`	LIFO	深度优先、回溯
`PriorityQueue`	最小堆，元素需可比较	按优先级调度

task_done() 与 join() 是一对：每次 get() 后必须调一次，

import queue, threading

q: queue.Queue = queue.Queue(maxsize=100)

def worker(n):
    while True:
        item = q.get()
        try:
            handle(item)
        finally:
            q.task_done()       # 必须放 finally，异常也要计数

for i in range(4):
    threading.Thread(target=worker, args=(i,), daemon=True).start()

for item in produce():
    q.put(item)
q.join()                        # 等所有 task_done

maxsize 是关键参数：不设上限（maxsize=0）时生产者不会阻塞，内存随消费速度差线性上涨，直到 OOM。有界队列 + 背压是唯一稳妥的生产者消费者形态。

4. 生产者消费者标准骨架

完整版还要处理"停止信号"和"异常传播"：

import queue, threading
from dataclasses import dataclass

@dataclass
class Stop: pass

def consumer(q, out, errors):
    while True:
        item = q.get()
        try:
            if isinstance(item, Stop):
                return
            out.append(transform(item))
        except Exception as e:
            errors.append(e)
        finally:
            q.task_done()

q, out, errors = queue.Queue(maxsize=64), [], []
workers = [threading.Thread(target=consumer, args=(q, out, errors)) for _ in range(4)]
[w.start() for w in workers]

for x in source():
    q.put(x)
for _ in workers:
    q.put(Stop())          # 每个消费者一个哨兵

for w in workers:
    w.join()
if errors:
    raise errors[0]

三个细节：哨兵数量等于消费者数；task_done() 放 finally，否则异常后 join 永不返回；list.append 在 CPython 里原子，但"读-改-写"不是，复杂结构要加锁。

5. concurrent.futures：两种执行器

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

with ProcessPoolExecutor(max_workers=4) as ex:
    futures = {ex.submit(cpu_task, x): x for x in data}
    for fut in as_completed(futures):
        x = futures[fut]
        try:
            print(x, fut.result())
        except Exception as e:
            print("failed", x, e)

API	行为
`submit`	立即返回 `Future`，不阻塞
`map`	按输入顺序返回迭代器，异常在迭代时抛
`as_completed`	完成即产出，顺序不确定，适合"谁先好先用"
`wait(..., return_when=FIRST_EXCEPTION)`	首个异常即返回，其余继续跑

ProcessPoolExecutor 与线程池的差别不止并行：提交的函数与参数必须可 pickle，lambda、闭包、模块级不可导入对象都会抛 PicklingError。子进程不共享内存，改全局变量无效。

6. 多进程通信与序列化开销

方式	语义	代价
`multiprocessing.Queue`	进程安全队列，内部走管道 + 锁	每次 put/get 都 pickle
`Pipe`	双端连接，`send`/`recv`	更快，但只适合一对一
`Value`/`Array`	共享内存 + 锁	只支持简单类型，无 pickle
`shared_memory`	真正的共享内存块	手动管理生命周期与同步

from multiprocessing import Process, Queue

def child(q):
    q.put(sum(range(10**6)))

q = Queue()
p = Process(target=child, args=(q,))
p.start()
print(q.get())
p.join()

序列化是常被忽略的成本：传一个 10 MB 的 DataFrame，pickle 往返耗时可能超过计算本身。批量传输通常比换通信方式更有效。

7. asyncio 原语与阻塞陷阱

原语	用途
`asyncio.gather`	并发等待多个协程，`return_exceptions=True` 时不中断
`asyncio.Semaphore`	限制并发数（爬虫、DB 连接）
`asyncio.Lock`	协程间互斥（不能跨线程）
`asyncio.Queue`	协程间生产消费，`maxsize` 同样重要
`asyncio.wait_for`	超时控制

import asyncio

sem = asyncio.Semaphore(10)

async def fetch(url):
    async with sem:
        await asyncio.sleep(0)          # 让出控制权
        return await client.get(url)

async def main(urls):
    results = await asyncio.gather(*(fetch(u) for u in urls), return_exceptions=True)
    return [r for r in results if not isinstance(r, Exception)]

致命错误是在协程里调阻塞函数：time.sleep(1)、同步 requests.get、同步 DB 驱动都会卡住整个事件循环。换 asyncio.sleep，或把同步库用 asyncio.to_thread 包一层。

asyncio.Lock 不能跨线程用，也不能在同步上下文里 await。它保护的是"协程切换点之间的临界区"，不是线程安全。

8. 超时与取消

场景	手段	限制
线程等待锁	`lock.acquire(timeout=1)`	只能超时，不能强杀
线程执行中	无官方手段	只能协作式：`Event` 标志位轮询
线程池任务	`future.cancel()`	只能取消"未开始"的任务
协程	`task.cancel()` + `CancelledError`	取消是协作的，阻塞调用不响应
协程整体	`asyncio.wait_for(coro, 5)`	超时后抛 `TimeoutError`

threading 没有强制终止：Thread.terminate 不存在，_thread.interrupt_main 只影响主线程。想在超时后停止工作线程，只能让工作函数周期性检查一个 threading.Event。

asyncio 的取消靠往协程里抛 CancelledError：

async def guarded():
    try:
        await long_running()
    except asyncio.CancelledError:
        cleanup()
        raise                 # 必须重新抛出，别吞

吞掉 CancelledError 会让 task.cancel() 失效，任务变成"取消不掉"的僵尸。3.8 之后它继承 BaseException，正是为了让 except Exception 不要误捕获。

9. 死锁的四个必要条件在 Python 里的样子

条件	Python 里的具体形态
互斥	`Lock` 独占，别人拿不到
持有并等待	拿着 A 去 `acquire(B)`
不可剥夺	`Lock` 没有超时抢夺，只能等
循环等待	线程 1 拿 A 等 B，线程 2 拿 B 等 A

破掉任意一个即可。实践中最便宜的是破"循环等待"：给所有锁规定全局获取顺序，任何线程都按同一顺序加锁。次便宜的是破"持有并等待"：一次性获取所有锁，或用 acquire(timeout=...) 失败即释放重试。

检测：faulthandler.dump_traceback_later(30, exit=True) 卡死时打印所有线程栈，或 py-spy dump --pid <pid> 看谁等在哪把锁。

10. 排错表

现象	原因	处理
线程池比单线程还慢	CPU 密集 + GIL	换 `ProcessPoolExecutor`
协程全部卡住	协程里有阻塞调用	换异步 API 或 `to_thread`
内存随运行时间上涨	无界 `Queue`	设 `maxsize` 并做背压
`join()` 永不返回	漏 `task_done`，或线程非 daemon 未退出	补 `task_done`，或用哨兵结束
程序卡死但 CPU 为 0	同线程重复 `acquire` 非重入锁	换 `RLock`
计数结果偏小	共享可变对象未加锁	`with lock` 包住读改写

11. 边界

需求	首选	备选
万级并发 IO	`asyncio`	线程池 + 大 `maxsize`
数值计算	`multiprocessing` / numpy	C 扩展
简单并行批处理	`concurrent.futures`	手工线程
需要共享大数组	`shared_memory`	`Array`
跨进程限流	外部存储（Redis）	`Semaphore` 只在单进程内有效

最后一条：两种 Semaphore 都是进程内的，多实例部署时限的是"每进程 10 个"，总并发是 10 × 实例数。

12. 自测

默认构建下，线程池为什么不能加速 CPU 密集任务？
Lock 在同一线程重复获取会怎样？该换成什么？
queue.join() 依赖哪个方法被调用？漏掉会怎样？
ProcessPoolExecutor 提交任务时，哪些对象会导致 PicklingError？
为什么不能在协程里调用 time.sleep？
CancelledError 被 except Exception 捕获后吞掉，会发生什么？