第 12 章 · 多线程与锁
本章目标：理解 GIL 对 CPU 密集任务的影响；使用 threading 创建线程；掌握 Lock / RLock / Condition 与 queue.Queue 线程安全队列；能编写 I/O 密集 并发脚本（如多 URL 探测、多文件解析）；知道线程与进程的选型边界（进程见 ch13，协程见 ch14）。
学时建议：4～5 小时（含 2 小时实验）
前置：ch08 文件与异常；ch11 requests（I/O 场景）；ch07 上下文管理器。
12.1 为什么需要多线程
| 任务类型 | 瓶颈 | 典型手段 |
|---|---|---|
| I/O 密集 | 等网络/磁盘 | 多线程、协程 |
| CPU 密集 | 算力 | 多进程、C 扩展、NumPy |
| 混合 | 视热点 | 先 profile 再选型 |
shop-demo 场景：同时拉取 10 个商品详情 API、并行解析 5 个访问日志文件——线程在 等待 I/O 时可切换，提高吞吐。
单线程： [请求1等待][请求2等待][请求3等待] → 串行累加延迟
多线程： [请求1等待]
[请求2等待] → 等待重叠，总时间接近最慢那个
[请求3等待]12.2 GIL（全局解释器锁）
CPython 中，同一时刻只有一个线程执行 Python 字节码。
| 影响 | 说明 |
|---|---|
| CPU 密集 + 多线程 | 几乎无加速，还可能更慢 |
| I/O 密集 + 多线程 | I/O 阻塞时释放 GIL，有效 |
| 计算密集 | 用 multiprocessing（ch13） |
结论：本章线程适合 I/O；算排序、算哈希请用下一章进程池。
12.3 创建线程
import threading
import time
def fetch_product(sku: str) -> None:
print(f"[{threading.current_thread().name}] fetching {sku}")
time.sleep(0.5) # 模拟网络延迟
print(f"[{threading.current_thread().name}] done {sku}")
threads = [
threading.Thread(target=fetch_product, args=(f"SKU-{i}",), name=f"worker-{i}")
for i in range(3)
]
for t in threads:
t.start()
for t in threads:
t.join() # 等待全部结束
print("all finished")| API | 作用 |
|---|---|
Thread(target=fn, args=(...)) | 创建线程 |
start() | 启动 |
join() | 阻塞直到线程结束 |
threading.current_thread() | 当前线程对象 |
12.4 Lock 与 RLock
多线程写共享变量会竞态：
# 危险示例：count 可能小于 100000
count = 0
def add_many():
global count
for _ in range(100_000):
count += 1Lock（互斥锁）：
import threading
count = 0
lock = threading.Lock()
def add_many():
global count
for _ in range(100_000):
with lock: # 等价 acquire / release
count += 1
threads = [threading.Thread(target=add_many) for _ in range(4)]
for t in threads:
t.start()
for t in threads:
t.join()
print(count) # 400000| 类型 | 说明 |
|---|---|
Lock | 同一时刻仅一个线程持有 |
RLock | 可重入，同一线程可多次 acquire |
Condition | 等待/通知，配合锁做生产者消费者 |
RLock 示例（递归函数内需同一线程重复加锁）：
rlock = threading.RLock()
def outer():
with rlock:
inner()
def inner():
with rlock:
print("ok")12.5 queue.Queue：线程安全队列
不要用 list 在线程间裸传数据；用 queue.Queue：
import queue
import threading
import time
task_q: queue.Queue[str] = queue.Queue()
result_q: queue.Queue[int] = queue.Queue()