Python 并发编程：threading、multiprocessing、concurrent.futures 与 GIL

Python 的并发比多数语言更绕，因为有个 GIL 横在中间：同样是「开线程」，在 Java 里能并行跑满多核，在 CPython 里却常常只能并发不能并行。于是 Python 的并发分成了三条路线，各自服务的场景泾渭分明：

threading --多线程。受 GIL 限制，同一时刻只有一个线程执行 Python 字节码，但 IO 阻塞时会释放 GIL，适合 IO 密集任务。
multiprocessing --多进程。每个进程有独立的解释器和 GIL，能真正用满多核，适合 CPU 密集任务，代价是进程创建和通信的开销。
asyncio --协程。单线程 + 事件循环，并发量极高但只适合 IO 等待，详见 asyncio 一文。

本文先讲清楚 GIL 这个「为什么」的前提，再分别拆解 threading 和 multiprocessing，最后用 concurrent.futures 把两者统一成一个接口。理解这三条路线的边界，比记住任何 API 都重要。示例以 Python 3.12 为基准。

GIL：理解 Python 并发的第一道门槛

GIL（Global Interpreter Lock，全局解释器锁）是 CPython 实现层面的一个互斥锁，保证同一时刻只有一个线程在执行 Python 字节码。它不是语言规范的一部分，而是 CPython 的实现选择–Jython、IronPython 没有 GIL。

为什么要有 GIL

CPython 的内存管理基于引用计数：每个对象有一个引用计数，指向它的引用增减时同步更新，归零时回收。多线程同时修改同一对象的引用计数需要加锁保护。早期设计者选择了简单粗暴的方案：用一把全局锁保护整个解释器的状态。代价是「多线程无法并行执行 Python 代码」，收益是「解释器内部数据结构不用处处加细粒度锁，C 扩展写起来也简单」。在单核时代这笔交易很划算。

GIL 何时释放

GIL 不是「一个线程永远霸占」，而是按规则切换：

IO 阻塞时主动释放：read/write/sleep/recv 等系统调用前释放 GIL，让其他线程跑。所以 IO 密集的多线程是能并发的。
按时间片切换：3.2+ 采用「切换间隔」策略，默认每 5ms（sys.getswitchinterval()）尝试切换一次。

用一段代码直观感受 GIL 对 CPU 密集任务的影响：

import threading, time

def countdown(n: int) -> None:
    while n > 0:
        n -= 1

# 两线程跑纯计算，GIL 不释放
start = time.perf_counter()
t1 = threading.Thread(target=countdown, args=(50_000_000,))
t2 = threading.Thread(target=countdown, args=(50_000_000,))
t1.start(); t2.start(); t1.join(); t2.join()
print(f"两线程: {time.perf_counter()-start:.2f}s")   # 和串行差不多甚至更慢

这就是「GIL 让多线程在 CPU 密集任务上失效」的直接证据。

CPU 密集 vs IO 密集

这张判断决定了你该用哪条路线：

任务特征	GIL 影响	推荐方案
CPU 密集（数值计算、压缩、加密）	致命，多线程无法并行	`multiprocessing` 或 C 扩展
IO 密集（网络、文件、数据库）	几乎无影响，阻塞时释放 GIL	`threading` 或 `asyncio`
混合型	看瓶颈在哪	asyncio 主调度 + 进程池跑计算

绕过 GIL 的两条路

free-threaded 模式（3.13 实验性）：PEP 703 去掉了 GIL，让多线程真正并行，代价是单线程性能略降、C 扩展需重新适配。目前仍处实验阶段，多数第三方库尚未支持，生产环境应谨慎。在可预见的未来，「CPU 密集用多进程」仍是稳妥选择。
C 扩展主动释放 GIL：NumPy、hashlib、zlib 等库在 C 层面的紧密循环中会主动释放 GIL，所以「用 NumPy 做矩阵运算」的多线程代码能并行–计算发生在 GIL 之外的 C 代码里。这也是数据科学场景下「多线程 + NumPy」有时能用满多核的原因。

threading：多线程与线程同步

threading 模块是 Python 的线程抽象。线程轻量（相比进程），共享进程内存，适合 IO 密集任务。

创建与等待线程

import threading, time

def worker(name: str, seconds: float) -> None:
    print(f"{name} 开始")
    time.sleep(seconds)          # 阻塞时释放 GIL，其他线程可跑
    print(f"{name} 结束")

t = threading.Thread(target=worker, args=("A", 1.0))
t.start()       # 启动
t.join()        # 等待结束

daemon=True 标记为守护线程，用于「主程序退出时不必等它」的后台任务（心跳、监控）。注意守护线程被强制终止时无法执行 finally，不要把必须完成的清理放在守护线程里。

线程安全与锁

多线程共享变量时，「读-改-写」不是原子的：

import threading

counter = 0
def increment(n: int) -> None:
    global counter
    for _ in range(n):
        counter += 1      # 字节码层面是 LOAD/INPLACE_ADD/STORE 三条指令，可能丢失更新

threads = [threading.Thread(target=increment, args=(100_000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)            # 期望 400000，实际经常小于 400000

用 Lock 保护：

lock = threading.Lock()
def increment(n: int) -> None:
    global counter
    for _ in range(n):
        with lock:           # 获取锁，离开 with 自动释放
            counter += 1

Lock 是不可重入锁，同一线程再次获取会死锁；RLock（可重入锁）允许同一线程多次获取，适合「锁内调用的函数也用到同一把锁」的递归或嵌套场景，代价是略慢。不确定时优先 Lock。

线程局部存储 `local`

不想加锁又想让每个线程有独立状态时，用 threading.local：

1
2
3

local_data = threading.local()
def worker():
    local_data.value = threading.current_thread().name   # 每个线程看到自己的 value

local_data 在每个线程里是独立的存储空间，互不干扰。这是数据库连接池、请求上下文等「每线程一份」场景的基础。

`Event` 与 `Condition`：线程间通知

event = threading.Event()
def waiter():
    event.wait()          # 阻塞直到 event.set()
    print("收到信号")

t = threading.Thread(target=waiter); t.start()
threading.Timer(1.0, event.set).start()   # 1s 后发出信号
t.join()

Event 用于「一个线程通知其他线程某事已发生」，wait(timeout) 可带超时。Condition 是 Lock + wait/notify 的组合，是经典生产者-消费者的实现方式。不过实际工程中更推荐直接用 queue.Queue，它内部就是用 Condition 实现的线程安全队列，省去手写同步的麻烦。

`queue.Queue`：线程安全的队列

import threading, queue

q = queue.Queue(maxsize=10)

def producer():
    for i in range(20):
        q.put(i)          # 队列满时阻塞
    q.put(None)           # 哨兵

def consumer():
    while True:
        item = q.get()    # 队列空时阻塞
        if item is None:
            break
        print(item)
        q.task_done()

threads = [threading.Thread(target=consumer) for _ in range(3)]
for t in threads: t.start()
producer()
for t in threads: t.join()

Queue 把锁、Condition、阻塞语义封装好了，是多线程下生产者-消费者的首选。还有 LifoQueue（栈）和 PriorityQueue（优先级队列）两个变体。

multiprocessing：多进程与真正的并行

GIL 让多线程无法并行执行 Python 代码，绕过的直接办法是开多个进程–每个进程有自己的解释器和 GIL，能真正跑在不同的 CPU 核上。

创建进程

import multiprocessing, time

def worker(name: str, seconds: float) -> None:
    print(f"{name} 开始, pid={multiprocessing.current_process().pid}")
    time.sleep(seconds)

if __name__ == "__main__":    # 这层保护必不可少，见后文「启动方式」
    p = multiprocessing.Process(target=worker, args=("A", 1.0))
    p.start()
    p.join()

接口和 threading.Thread 几乎一致，但底层是 fork/spawn 出一个新进程。回到前面 threading 失效的例子，换成多进程：

if __name__ == "__main__":
    start = time.perf_counter()
    procs = [multiprocessing.Process(target=countdown, args=(50_000_000,)) for _ in range(2)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(f"两进程: {time.perf_counter()-start:.2f}s")   # 约为串行的一半

两个进程跑在不同核上，耗时约为串行的一半–这才是真正的并行加速。

进程间通信

进程不共享内存，数据交换必须走 IPC。multiprocessing 提供 Queue 和 Pipe：

def producer(q):
    q.put("hello")
    q.put(None)

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"收到: {item}")

if __name__ == "__main__":
    q = multiprocessing.Queue()
    p1 = multiprocessing.Process(target=producer, args=(q,))
    p2 = multiprocessing.Process(target=consumer, args=(q,))
    p1.start(); p2.start(); p1.join(); p2.join()

multiprocessing.Queue 接口和 queue.Queue 一致，但底层用管道和 pickle 序列化传输，所以放入队列的对象必须可 pickle（函数、闭包、lambda 在某些启动方式下不行）。Pipe 是点对点的双向管道，性能略高，但只适合两个进程之间。

共享内存

传输大数组时序列化开销可观，Value/Array 在共享内存里存 C 类型：

import multiprocessing as mp

def worker(counter):
    for _ in range(10000):
        with counter.get_lock():
            counter.value += 1

if __name__ == "__main__":
    counter = mp.Value("i", 0)    # 共享的 C 类型整数，自带一把锁
    procs = [mp.Process(target=worker, args=(counter,)) for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(counter.value)          # 40000

对大型数值数组，multiprocessing.shared_memory.SharedMemory（3.8+）配合 NumPy 能避免拷贝，是高性能计算场景的选择。

启动方式：fork / spawn / forkserver

方式	行为	默认平台	特点
`fork`	复制父进程的整个内存空间	Linux（3.14 前）	快，但继承父进程状态可能导致死锁
`spawn`	启动全新进程，重新导入模块	Windows、macOS	慢，但干净安全
`forkserver`	预先 fork 一个服务进程，按需 fork	可选	结合两者优点

macOS 从 3.8 起默认 spawn，Linux 仍默认 fork（3.14 计划改为 forkserver）。fork 在父进程持有多线程时是不安全的–子进程只会复制调用了 fork 的那个线程，其他线程「消失」了，它们持有的锁永远不会再被释放，导致死锁。所以「主程序已经起了线程池就不要用 fork」是一条铁律。

`Pool`：进程池

手动管理进程很繁琐，Pool 提供了池化的便捷接口：

def square(x): return x * x

if __name__ == "__main__":
    with multiprocessing.Pool(processes=4) as pool:
        print(pool.map(square, range(10)))   # 阻塞，结果按顺序返回
        print(pool.map_async(square, range(10)).get(timeout=5))  # 非阻塞

Pool.map 自动把任务分发给池里的进程，适合「一批独立的 CPU 任务」。apply/apply_async 用于单个任务。

concurrent.futures：统一的线程池与进程池接口

threading 和 multiprocessing 的 API 各不相同，concurrent.futures 在两者之上提供了一致的「提交任务 + 获取 Future」接口，是日常并发最推荐的高层抽象。

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def fetch(url: str) -> str: ...   # IO 任务用线程池，CPU 任务用进程池

# IO 密集：线程池
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(fetch, urls))

# CPU 密集：进程池（默认 worker 数 = CPU 核数）
with ProcessPoolExecutor() as executor:
    results = list(executor.map(heavy_compute, data))

两者接口完全一致，只差类名。max_workers 控制并发数：线程池通常设为 IO 等待倍数，进程池设为 CPU 核数。

`submit` 与 `as_completed`

map 适合「同函数作用于一组输入」。更灵活的是 submit，返回 Future，可单独等待、查询状态：

from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=10) as executor:
    future = executor.submit(fetch, "https://example.com")
    print(future.result(timeout=10))   # 阻塞直到完成或超时
    print(future.exception())          # 任务抛异常时拿到异常对象

    # 谁先完成谁先处理
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            print(f"{url} -> {future.result()}")
        except Exception as e:
            print(f"{url} 失败: {e}")

用法	适合场景	结果顺序	异常处理
`executor.map(fn, iter)`	同质任务、要全部结果	按输入顺序	任一异常会抛出，影响整体
`submit` + `as_completed`	按完成顺序处理、要容错	按完成顺序	每个 Future 单独 `result()` 取异常

concurrent.futures 的价值在于把任务分片、提交、收集、异常处理用很少的代码表达清楚，且线程池和进程池可以无缝切换（只改一个类名）。

选型决策树

任务是 IO 等待为主吗？
├─ 是
│  ├─ 并发量巨大（数千连接）        -> asyncio
│  ├─ 中等并发、用同步库更顺手        -> threading / ThreadPoolExecutor
│  └─ 需要精确控制协程生命周期        -> asyncio
└─ 否（CPU 计算）
   ├─ 计算可切片、想用满多核          -> multiprocessing / ProcessPoolExecutor
   ├─ 计算在 C 扩展里（NumPy 等）     -> threading 即可（C 层会释放 GIL）
   └─ 单个任务很重                   -> 单进程 + C 扩展 / Cython

几条经验法则：

IO 密集 + 大量连接：asyncio 是最优解，单线程撑几万连接。
IO 密集 + 任务不多 + 想用 requests 这类同步库：ThreadPoolExecutor 最省事。
CPU 密集 + 想要并行加速：ProcessPoolExecutor，任务粒度别太细（进程创建和通信有开销）。
不确定：先用 concurrent.futures 写一版，线程池和进程池只改一个类名就能切换，方便 benchmark。

陷阱

线程 + fork 导致死锁–fork 只复制当前线程，父进程里其他线程持有的锁会变成「永远占用」的孤儿锁。解法：用 spawn 启动方式，或确保 fork 时不持有任何锁。macOS 默认 spawn 正是为了避免这类问题。

进程池里函数必须可 pickle–进程池要把函数和参数序列化传给子进程。顶层函数、模块级函数没问题；lambda、局部函数、某些闭包和实例方法不行，遇到 PicklingError 时把函数提到模块顶层。

任务粒度太细–进程池每提交一个任务都要序列化参数、跨进程传输、序列化结果返回。如果单个任务只算 x * x，通信开销远大于计算，多进程反而比串行慢几十倍。把数据切片成粗粒度的块再提交。

混用 threading.Lock 与 asyncio.Lock–协程和线程的锁不能互换：threading.Lock 的 acquire 是阻塞调用，会卡住事件循环；asyncio.Lock 的 acquire 是协程方法，在线程里没法 with。详见 asyncio 文章的陷阱一节。

以为多线程一定能加速–CPU 密集 + 多线程 = 浪费线程切换开销。先确认任务类型，再选路线。

忘记 if __name__ == "__main__"–在 spawn 启动方式下，子进程会重新导入主模块。如果创建进程的代码不在保护下，子进程导入时又会执行一遍，导致无限递归创建子进程。

总结

Python 并发的全貌可以浓缩成一张表：

路线	适合	并行？	内存共享	典型 API
`threading`	IO 密集、少量并发	否（GIL）	是	`Thread`、`Lock`、`queue.Queue`
`asyncio`	IO 密集、海量连接	否（单线程）	是（协程）	`async def`、`TaskGroup`
`multiprocessing`	CPU 密集	是	否（需 IPC）	`Process`、`Queue`、`Pool`
`concurrent.futures`	线程/进程池统一接口	取决于池	取决于池	`ThreadPoolExecutor`、`ProcessPoolExecutor`

理解 Python 并发的关键，是先回答两个问题：瓶颈是 IO 还是 CPU？需要并行还是并发？ 前者决定用线程/协程还是进程，后者决定能否绕开 GIL。GIL 让「多线程做 CPU 计算」失效，但「多线程做 IO 等待」依然有效；多进程是绕过 GIL 的正路，但任务粒度要够粗；concurrent.futures 把线程池和进程池统一成接口，是日常并发最省心的选择。

最后一条提醒：并发是手段，不是目的。如果单线程串行代码已经能在合理时间内跑完，引入并发只会增加复杂度、调试难度和出错可能。先确认性能瓶颈真实存在、确认瓶颈类型，再决定上哪条路线。多数脚本和中小服务，串行代码就是最好的选择。