知道了 GIL 的脾气，这一篇讲怎么真刀真枪地用线程和进程：锁住共享状态、跑满多核、在进程间传递数据，以及 Windows 上那些会让你踩坑的细节。
你将学到
threading的基础用法，以及为什么日常更该用ThreadPoolExecutor- 用
Lock/RLock/Event/Queue解决共享状态，而不是用sleep碰运气 - 守护线程是什么、什么时候会被直接掐死
multiprocessing与进程池：为什么它绕得过 GIL，以及 IPC 的三种姿势fork与spawn的区别，以及 Windows 上的两条铁律- 一段能跑的实测，对比 CPU 密集用多进程、IO 密集用多线程
前置知识
建议先读 06 - GIL 真相与并发模型选择，本篇的所有选型都建立在它的结论上。
一、threading 基础：从裸线程到线程池
最朴素的写法是 Thread(target=..., args=...) + start() + join()。裸线程的痛点很明显：手动管理启停、拿不到返回值、异常会被吞。日常工作请直接用 concurrent.futures.ThreadPoolExecutor，池子会自动维护线程生命周期：
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def fetch(name, delay):
time.sleep(delay)
return f"{name}:ok"
# 方式 1：map，保持输入顺序返回结果
with ThreadPoolExecutor(max_workers=5) as pool:
print(list(pool.map(fetch, ["a", "b", "c"], [0.3, 0.2, 0.1])))
# 输出: ['a:ok', 'b:ok', 'c:ok']（顺序与输入一致）
# 方式 2：submit + as_completed，谁先完成先处理（池子 with 退出时自动 shutdown）
with ThreadPoolExecutor(max_workers=5) as pool:
for fut in as_completed([pool.submit(fetch, f"t{i}", 0.1 * (i + 1)) for i in range(3)]):
print(fut.result()) # 输出: t0:ok / t1:ok / t2:ok（按完成先后）
max_workers 怎么定？ IO 密集任务里它是"并发上限"，可远大于 CPU 核数（几十到几百，取决于下游能扛多少）；CPU 密集任务多线程没意义，见第六节。
二、共享状态：锁、事件、队列
Lock / RLock
import threading
counter = 0
lock = threading.Lock()
def incr():
global counter
for _ in range(100_000):
with lock: # 上下文管理器自动 acquire/release
counter += 1 # 读-改-写整体原子，结果不会丢
ts = [threading.Thread(target=incr) for _ in range(4)]
for t in ts: t.start()
for t in ts: t.join()
print(counter) # 输出: 400000（这次一定对）
# RLock（可重入锁）允许同一线程多次获取，适合递归或方法间互相调用：
rlock = threading.RLock()
def outer():
with rlock: inner() # 同一线程再次拿锁，RLock 不会死锁
def inner():
with rlock: pass
outer()
# 注意：普通 Lock 非重入，同一线程重复 acquire() 会把自己锁死
Event：线程间的红绿灯
Event 适合"一个线程发信号、多个线程等待"的场景，别用 sleep 轮询。
import threading, time
ready = threading.Event() # 初始为"未置位"
def producer():
time.sleep(0.5)
print("生产者：数据准备好了")
ready.set() # 置位，唤醒所有等待者
def consumer(name):
ready.wait() # 阻塞，直到 set() 被调用；可加 timeout
print(f"{name}：开始消费")
threading.Thread(target=producer).start()
for i in range(2):
threading.Thread(target=consumer, args=(f"消费者{i}",)).start()
# 输出: 生产者：数据准备好了 / 消费者0：开始消费 / 消费者1：开始消费
Queue：最省心的线程安全队列
queue.Queue 内部自带锁，是生产者-消费者模型的标准答案，比手动加锁优雅得多。
import queue, threading
q = queue.Queue(maxsize=10)
def producer():
for i in range(5):
q.put(i) # 队列满则阻塞
q.put(None) # 哨兵，通知消费者结束
def consumer():
while True:
item = q.get() # 队列空则阻塞
if item is None:
break
print(f"消费: {item}")
q.task_done()
p = threading.Thread(target=producer); c = threading.Thread(target=consumer)
p.start(); c.start(); p.join(); c.join()
# 输出: 消费: 0 / 消费: 1 / 消费: 2 / 消费: 3 / 消费: 4
三、守护线程：会被"当场处决"
import threading, time
def background():
while True:
time.sleep(1); print("后台心跳...")
t = threading.Thread(target=background, daemon=True)
t.start()
time.sleep(2.5)
print("主线程结束")
# 输出: 后台心跳... / 后台心跳... / 主线程结束
# 进程随即退出，后台线程被直接终止——不会执行任何收尾逻辑
守护线程的语义是"主线程一退出就陪葬"。 日志刷新、连接归还这类需要收尾的工作，不要放在守护线程里；纯心跳、纯监控则可以设成守护线程，省得手动 join。
四、multiprocessing：绕开 GIL
多进程的核心思想：每个进程有自己独立的解释器和独立的 GIL，因此能同时占满多个 CPU 核。 代价是进程间不能共享内存（默认），数据要序列化后拷贝。进程池 ProcessPoolExecutor 让并发管理自动化，并支持 map 返回结果：
from multiprocessing import Process, Queue
from concurrent.futures import ProcessPoolExecutor
def cpu_task(q, n):
q.put(sum(i * i for i in range(n))) # 结果通过 Queue 送回主进程
def square(x):
return x * x
if __name__ == "__main__": # ⚠️ Windows 必须写，见第六节
q = Queue()
procs = [Process(target=cpu_task, args=(q, 2_000_000)) for _ in range(2)]
for p in procs: p.start()
for p in procs: p.join()
print([q.get() for _ in procs]) # 输出: [两个整数]
with ProcessPoolExecutor(max_workers=4) as pool:
print(list(pool.map(square, range(5)))) # 输出: [0, 1, 4, 9, 16]
进程间通信（IPC）三种姿势
from multiprocessing import Process, Pipe
# 1) Queue：多生产者多消费者，简单可靠（内部有锁，会做序列化）
# 2) Pipe：两端点一对一，速度更快，适合固定两个进程通信
# 3) 共享内存（shared_memory）：大数据量场景，省去序列化拷贝，需自己同步
def child_job(conn):
conn.send("hi"); conn.close() # 模块级函数才可 pickle
if __name__ == "__main__":
parent, child = Pipe() # 返回一对 Connection
p = Process(target=child_job, args=(child,))
p.start()
print(parent.recv()) # 输出: hi
p.join()
经验法则：小数据用 Queue/Pipe，大数据（几十 MB 以上、numpy 数组）用共享内存，否则光是序列化就能吃掉并行省下的时间。
五、fork vs spawn：启动方式决定一切
import multiprocessing as mp
# fork：直接复制父进程内存（Linux 默认），启动快，
# 但多线程程序 + fork 可能死锁（子进程只 fork 了当前线程）
# spawn：全新启动一个解释器，重新 import 模块，干净但慢
print(mp.get_start_method()) # Linux: fork / macOS、Windows: spawn
# 显式指定（3.14 起 fork 在 POSIX 被标为不安全，建议统一用 spawn）
ctx = mp.get_context("spawn")
ctx.Process(target=print, args=("hi",)).start() # 输出: hi
fork |
spawn |
|
|---|---|---|
| 平台默认 | Linux | Windows / macOS |
| 启动速度 | 快 | 慢（要重新导入） |
| 内存 | 写时复制，省 | 独立，费 |
| 线程安全 | 有死锁风险 | 安全 |
| 全局变量 | 子进程继承 | 不继承，需参数传入 |
六、Windows / macOS 上的两条铁律
既然默认是 spawn，就有两个必须遵守的规则：
from concurrent.futures import ProcessPoolExecutor
# 铁律 1：入口必须用 if __name__ == "__main__": 保护。原因：spawn 会重新 import
# 主模块，若进程创建代码在模块顶层，子进程 import 时会再次执行到它 -> 无限递归。
# ❌ 模块顶层直接 with ProcessPoolExecutor() as pool: pool.map(work, range(5))
def work(x):
return x + 1
# 铁律 2：传给子进程的东西必须能被 pickle 序列化
# ❌ pool.map(lambda x: x + 1, range(5)) # 报错：Can't pickle <lambda>
# ✅ 正确：放进 main 守卫 + 用模块顶层定义的普通函数
if __name__ == "__main__":
with ProcessPoolExecutor() as pool:
print(list(pool.map(work, range(5)))) # 输出: [1, 2, 3, 4, 5]
除这两条外还有：子进程里改全局变量父进程看不到；进程池里 print 输出可能交错；multiprocessing 在 Jupyter / 交互式解释器里经常不工作（拿不到可 import 的主模块）。
七、实测：多进程 vs 多线程
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
def cpu_task(n): # CPU 密集：纯计算
return sum(i * i for i in range(n))
def io_task(sec): # IO 密集：纯等待
time.sleep(sec)
def bench(fn, label):
start = time.perf_counter()
fn()
print(f"{label}: {time.perf_counter() - start:.3f}s")
if __name__ == "__main__":
N, K = 4_000_000, 4
bench(lambda: [cpu_task(N) for _ in range(K)], "CPU 串行 ")
bench(lambda: list(ThreadPoolExecutor(4).map(cpu_task, [N] * K)), "CPU 多线程 ")
bench(lambda: list(ProcessPoolExecutor(4).map(cpu_task, [N] * K)), "CPU 多进程 ")
bench(lambda: [io_task(0.5) for _ in range(8)], "IO 串行 ")
bench(lambda: list(ThreadPoolExecutor(8).map(io_task, [0.5] * 8)), "IO 多线程 ")
# 参考输出（4 核机器，量级示意，非精确值）:
# CPU 串行 : 2.60s
# CPU 多线程 : 2.75s <- 抢 GIL，几乎没快，甚至更慢
# CPU 多进程 : 0.80s <- 约 3 倍加速，真并行
# IO 串行 : 4.001s
# IO 多线程 : 0.502s <- 8 倍并发，接近理论极限
一句话结论：CPU 密集选进程，IO 密集选线程。 反过来选，除了让代码更复杂，没有任何收益。
常见坑
# ❌ 坑 1：用 sleep 轮询代替 Event
while not data_ready:
time.sleep(0.01) # 浪费 CPU，还可能有肉眼可见的延迟
# ✅ 正解：用 Event 精确唤醒
event.wait() # 零轮询开销，事件一到立刻继续
# ❌ 坑 2：守护线程里做必须完成的工作
t = threading.Thread(target=lambda: write_log_to_disk(), daemon=True)
# 主线程一退出，日志可能只写了一半，文件损坏
# ✅ 正解：需要收尾的线程设成非守护，并显式 join：
# t = threading.Thread(target=write_log_to_disk); t.start(); t.join()
# ❌ 坑 3：多进程里改全局变量，指望父进程看到（内存不共享，只改了子进程副本）
g = 0
def bump():
global g
g = 100 # 父进程永远还是 0
# ✅ 正解：要共享结果用 Queue / Pipe 传回，或用 multiprocessing.Value("i", 0)
小结
- 日常用
ThreadPoolExecutor/ProcessPoolExecutor，自动管理生命周期，比裸Thread/Process省心。 - 共享状态优先用队列（自带锁）而非手动加锁；必须共享标量时用
Lock，递归场景用RLock。 - 守护线程 = 主线程退出即被处决，收尾工作（日志、连接归还）别放里面。
- 多进程能绕开 GIL 是因为每个进程有独立的解释器和 GIL；IPC 三件套
Queue通用、Pipe快、SharedMemory省拷贝。 - Windows/macOS 默认
spawn：必须写if __name__ == "__main__":，传入对象必须可 pickle；实测一句话：CPU 密集用多进程，IO 密集用多线程。
延伸阅读
- Python 官方文档：
concurrent.futures、multiprocessing、queue模块 - 《Python 并发编程实战》关于 IPC 与进程池调优的章节
文章回复
0 条公开回复