Python 与数据实践 · 第 18 章

线程与进程

到目前为止，我们写的所有程序都活在单线程世界里：代码从上往下、一件接一件地执行。可现实中的任务常常不是这样——你可能要给几百张照片批量生成缩略图，要同时调用十几个外部接口，要把一段视频逐帧处理。单线程只能老老实实排队，一…

约 27 分钟10,837 字18 个代码示例

到目前为止，我们写的所有程序都活在单线程世界里：代码从上往下、一件接一件地执行。可现实中的任务常常不是这样——你可能要给几百张照片批量生成缩略图，要同时调用十几个外部接口，要把一段视频逐帧处理。单线程只能老老实实排队，一件做完才轮到下一件。

想让程序"同时"做多件事，就要进入并发的世界。这条路有几条岔道：操作系统层面的进程与线程，以及 Python 在用户态玩出的花样——协程。本章先走前两条，把 threading 和 multiprocessing 两件工具用熟；协程背后的引擎——事件循环——留到下一章拆。

1. 先把任务分进两个筐

任何耗时的任务，大致都能归进两个筐里：

I/O 密集型：时间花在"等"上——等网络响应、等磁盘读写、等数据库返回。CPU 大部分时间其实在闲着；
CPU 密集型：时间花在"算"上——图像处理、数值计算、压缩加密。CPU 从头到尾满负荷。

任务属于哪一类，直接决定并发方案怎么选。先记一句话垫底：多线程擅长处理等 I/O 的任务，多进程擅长吃满 CPU 的任务，协程则是海量 I/O 任务的终极形态。本章末尾还有一张速查表，到时候回头验证这句话。

2. 进程、线程、协程：三位选手同场对比

并发三兄弟放在一起看，差异一目了然：

对比维度	🏭 进程 (Process)	🧵 线程 (Thread)	🍃 协程 (Coroutine)
由谁调度	操作系统（内核态）	操作系统（内核态）	用户态的事件循环
创建开销	最大：独立内存空间、独立文件描述符，创建要走 fork/clone	中等：共享进程内存，但要有自己的栈（MB 级）	极小：几 KB 的状态就够了
切换成本	最高：切页表、刷 TLB	高：要进内核态，几百个 CPU 周期	最低：纯用户态，几十个周期
数据共享	天然隔离，只能靠管道、socket 等 IPC	同一屋檐下共享内存，必须做同步	单线程内执行，天然安全
多核利用	✅ 能	✅ 能	❌ 单线程内不能
合适数量级	十个左右	几百上千	成千上万
典型场景	需要隔离风险的独立任务	CPU 计算、同步阻塞 I/O	大规模 I/O 并发

3. threading 上手：把函数扔进另一个线程

最简用法：Thread + start + join

import threading
import time


def compress_photo(name: str, seconds: int):
    print(f"图片 {name} 开始压缩")
    time.sleep(seconds)  # 假装在干活
    print(f"图片 {name} 压缩完毕")


t = threading.Thread(target=compress_photo, args=("sunset.jpg", 2))
t.start()  # 线程交给操作系统去跑
print("主线程该干嘛干嘛")  # 这行会立刻执行，不必等压缩完成
t.join()   # 需要结果了，再等它
print("主线程等到收工")

输出：

图片 sunset.jpg 开始压缩
主线程该干嘛干嘛
……（约 2 秒后）
图片 sunset.jpg 压缩完毕
主线程等到收工

三个动作各司其职：start() 把线程交给操作系统调度；join() 让当前线程原地等待，直到对方跑完；夹在中间那行输出的时机，证明了两条线确实是并行的。

面向对象写法：继承 Thread

import threading
import time


class PhotoWorker(threading.Thread):
    def __init__(self, filename: str, seconds: int):
        super().__init__(name=f"worker-{filename}")
        self._seconds = seconds

    def run(self) -> None:
        print(f"{self.name} 开工，耗时预算 {self._seconds} 秒")
        time.sleep(self._seconds)
        print(f"{self.name} 完工")


task = PhotoWorker("avatar.png", 1)
task.start()
task.join()

要点只有一个：把任务逻辑写进 run()，start() 会自动调用它。

守护线程：随主线程一起退场

普通线程会"拽住"进程——只要它没跑完，进程就不退出。守护线程（daemon）正相反：主线程一结束，它立刻被强制掐断。

import threading
import time


def heartbeat():
    while True:
        print("心跳：进程还活着")
        time.sleep(1)


monitor = threading.Thread(target=heartbeat, daemon=True)
monitor.start()

time.sleep(3)
print("主线程跑完了，守护线程被一并带走")

提醒：守护线程是"被掐断"，不是"善后退出"。写文件、提交事务这类需要完整收尾的操作，千万别放在守护线程里——它可能做到一半就被带走了。

4. 共享内存的代价：竞态条件

线程最方便的地方是大家共享同一块内存，最危险的地方也在这里。多个线程同时读写同一个变量，就可能踩出竞态条件（Race Condition）。

看一个秒杀场景：库存只剩 10 件，两位买家同时下单、各抢 8 件——

import threading
import time


class FlashStock:
    def __init__(self, count: int):
        self.count = count


def place_order(stock: FlashStock, quantity: int, buyer: str) -> bool:
    """下单——这一版带着竞态漏洞"""
    print(f"{buyer} 查库存：还剩 {stock.count} 件")

    # 漏洞所在：查库存和扣库存是两步，中间有空子
    if stock.count >= quantity:
        print(f"{buyer}：库存够，下单中……")

        # 模拟生成订单、走支付流程的耗时
        # 这 0.1 秒，足够另一个线程钻进来了
        time.sleep(0.1)

        stock.count -= quantity
        print(f"{buyer} ✓ 抢到 {quantity} 件！库存剩 {stock.count} 件")
        return True

    print(f"{buyer} ✗ 库存不足，手慢了")
    return False


stock = FlashStock(10)

buyer1 = threading.Thread(target=place_order, args=(stock, 8, "买家甲"))
buyer2 = threading.Thread(target=place_order, args=(stock, 8, "买家乙"))

print("=== 秒杀开始，两人同时下单 ===")
buyer1.start()
buyer2.start()
buyer1.join()
buyer2.join()

print("\n=== 结算 ===")
print(f"实际库存：{stock.count} 件")
print(f"本应剩下：{10 - 8} = 2 件")
print(f"实际卖出：{10 - stock.count} 件——超卖了！")

一次典型的运行结果：

=== 秒杀开始，两人同时下单 ===
买家甲 查库存：还剩 10 件
买家乙 查库存：还剩 10 件
买家甲：库存够，下单中……
买家乙：库存够，下单中……
买家甲 ✓ 抢到 8 件！库存剩 2 件
买家乙 ✓ 抢到 8 件！库存剩 -6 件

=== 结算 ===
实际库存：-6 件
本应剩下：2 件
实际卖出：16 件——超卖了！```

两个人都"看到"库存有 10 件，都判断充足，都扣了 8 件，最后库存成了 -6。

> 为什么会这样？`stock.count -= quantity` 看着是一行代码，落到 CPU 上是三条指令：**读出** count、**做减法**、**写回** count。线程切换完全可能发生在任意两条指令之间。更麻烦的是，"检查够不够"和"真的扣掉"这两个动作之间，本身就留着一条时间缝。

## 5. Lock：一次只放一个人进房间

`threading.Lock` 是最基础的互斥锁。它把一段临界区变成"单人房间"：谁拿到钥匙谁进，其他人一律在门口排队。用它堵住秒杀漏洞：

```python
import threading
import time


class FlashStock:
    def __init__(self, count: int):
        self.count = count
        self.lock = threading.Lock()


def place_order(stock: FlashStock, quantity: int, buyer: str) -> bool:
    """下单——用锁保证线程安全"""
    with stock.lock:  # 拿钥匙；拿不到的线程在此阻塞
        print(f"{buyer} 查库存：还剩 {stock.count} 件")

        if stock.count >= quantity:
            print(f"{buyer}：库存够，下单中……")
            time.sleep(0.1)
            stock.count -= quantity
            print(f"{buyer} ✓ 抢到 {quantity} 件！库存剩 {stock.count} 件")
            return True

        print(f"{buyer} ✗ 库存不足，手慢了")
        return False


stock = FlashStock(10)

buyer1 = threading.Thread(target=place_order, args=(stock, 8, "买家甲"))
buyer2 = threading.Thread(target=place_order, args=(stock, 8, "买家乙"))

print("=== 秒杀开始，两人同时下单 ===")
buyer1.start()
buyer2.start()
buyer1.join()
buyer2.join()

print("\n=== 结算 ===")
print(f"实际库存：{stock.count} 件")  # 2 件，正确

第二个买家进房间时，库存已经是 2，不够 8，只能悻悻而归。with 语句还保证了一件事：房间里的代码无论是正常走完还是中途抛异常，出门时一定把钥匙还回去。

用锁有个新手必踩的坑——自己锁自己：

import threading

lock = threading.Lock()

lock.acquire()
lock.acquire()  # 卡死：钥匙还在自己手里，却还想要一次

第一个 acquire 拿到锁还没释放，第二个 acquire 只能干等——可"释放"永远不会发生。这就是死锁的一种最简单形态。

6. RLock：同一扇门可以反复进

RLock（可重入锁）在内部记了一个数：同一个线程每 acquire 一次计数加一，每 release 一次减一，减到零才真正开门。于是"自己锁自己"不再成立——适合递归调用、或同一线程里嵌套加锁的场合：

import threading

rlock = threading.RLock()


def countdown(n: int):
    with rlock:
        if n > 0:
            print(f"倒数 {n}")
            countdown(n - 1)  # 同一个线程再次拿锁，放行 ✅


countdown(5)  # 顺利跑完

7. Semaphore：给房间挂个限流牌

Lock 是"一次一个人"，Semaphore（信号量）是"一次 N 个人"。比如机器只有 3 个核能干活，渲染任务却来了 10 个，就得控制同时在跑的最多 3 个：

import threading
import time

permits = threading.Semaphore(3)  # 3 个名额


def render_frame(no: int):
    with permits:  # 占一个名额；满了就在此等待
        print(f"第 {no} 号帧开始渲染")
        time.sleep(1)
        print(f"第 {no} 号帧渲染完成")


threads = [threading.Thread(target=render_frame, args=(i,)) for i in range(10)]

for t in threads:
    t.start()
for t in threads:
    t.join()

观察输出会发现：任意时刻最多 3 帧在渲染，一批结束，下一批立刻顶上。

8. Event：线程之间的发令枪

很多时候线程之间需要的不是"互斥"而是"通知"：一个线程等某个条件成立，另一个线程负责宣布它成立。Event 就是这把发令枪：

import threading
import time

gun = threading.Event()


def sprinter():
    print("选手：已就位，等枪响")
    gun.wait()  # 阻塞，直到事件被 set
    print("选手：枪响了，冲！")


def referee():
    print("裁判：各就各位——预备——")
    time.sleep(1)
    gun.set()  # 扣动扳机
    print("裁判：比赛开始")


runner = threading.Thread(target=sprinter)
judge = threading.Thread(target=referee)

runner.start()
judge.start()
runner.join()
judge.join()

wait() 是全体听枪，set() 是一声枪响——所有等在 wait() 上的线程同时被唤醒。

9. Queue：生产者和消费者之间的传送带

一个经典架构：一个线程生产数据，若干线程消费数据。要是自己拿列表加锁去实现，要操心的事非常多；标准库的 queue.Queue 直接把同步全做掉了：

import queue
import random
import threading
import time


def scanner(q: queue.Queue):
    """生产者：模拟扫描文件夹，不断发现新图片"""
    for i in range(10):
        filename = f"IMG_{i:03d}.jpg"
        q.put(filename)  # 队列满了会自动阻塞，等着消费者来取
        print(f"发现图片：{filename}")
        time.sleep(random.random())
    # 有几个消费者，就投几个"收工"信号
    q.put(None)
    q.put(None)


def compressor(name: str, q: queue.Queue):
    """消费者：从队列里取图片做压缩"""
    while True:
        filename = q.get()
        if filename is None:  # 收到收工信号
            q.task_done()
            break
        print(f"  {name} 正在压缩 {filename}")
        time.sleep(0.3)
        q.task_done()


q = queue.Queue(maxsize=5)  # 传送带长度有限，生产太快会被反压
threads = [
    threading.Thread(target=scanner, args=(q,)),
    threading.Thread(target=compressor, args=("压缩机1号", q)),
    threading.Thread(target=compressor, args=("压缩机2号", q)),
]

for t in threads:
    t.start()
for t in threads:
    t.join()

queue.Queue 内部已经处理好了加锁，put / get 尽管放心用；maxsize 给传送带限了长度，生产者跑得太快时会被自动"反压"。

10. 线程池：造线程不如租线程

线程的创建和销毁并不便宜，任务一多就该用池子——预先造好一批线程反复使用。concurrent.futures.ThreadPoolExecutor 就是现成的线程池：

from concurrent.futures import ThreadPoolExecutor
import time


def make_thumbnail(album: str) -> str:
    time.sleep(1)  # 模拟读原图、缩放、写盘
    return f"《{album}》的缩略图已生成"


albums = ["旅行相册", "毕业纪念", "产品图集", "宠物日常", "深夜食堂"]

with ThreadPoolExecutor(max_workers=3) as pool:
    # map：结果顺序与提交顺序严格一致
    for result in pool.map(make_thumbnail, albums):
        print(result)

    # submit：逐个提交，拿到 future 凭条，再凭条取结果
    futures = [pool.submit(make_thumbnail, album) for album in albums]
    for f in futures:
        print(f.result())

map 像内置的 map 一样按序返回结果；submit 返回的是 future 对象——先欠着，活干完了凭它取货。future 是个重要角色，第 20 章会专门请它出场。

11. 线程池的另一个身份：通往异步世界的接口

这里先剧透一点后面的内容：Python 的现代异步体系（协程 + 事件循环，从下一章开始讲）有条铁律——协程里绝不能出现阻塞调用。可现实总免不了要用同步阻塞的库：老牌数据库驱动、某些只提供同步接口的 SDK。这时候线程池就来救场：把阻塞函数扔到别的线程里跑，协程这边挂起等待，互不耽误：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor


def read_large_file() -> str:
    time.sleep(0.5)  # 同步读文件，会把所在线程按住
    return "大文件内容读取完毕"


async def main():
    # 方式一：asyncio.to_thread，最省事的写法
    text = await asyncio.to_thread(read_large_file)
    print(text)

    # 方式二：手动指定一个线程池
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        text = await loop.run_in_executor(pool, read_large_file)
        print(text)


asyncio.run(main())

两种写法效果一致：阻塞发生在别的线程里，事件循环所在的线程保持畅通。里面的原理，接下来几章会逐步展开。

12. threading.local：给每个线程一个储物柜

共享数据要加锁，那反过来——每个线程想有"只属于自己"的数据，怎么办？threading.local() 造出一排储物柜：各线程读写同名属性，互不可见、互不干扰：

import threading
import time

locker = threading.local()


def serve(window: str):
    locker.window = window   # 各写各的
    locker.handled = 0
    for _ in range(3):
        locker.handled += 1
        print(f"{locker.window}：已处理 {locker.handled} 个请求")
        time.sleep(0.1)


threads = [
    threading.Thread(target=serve, args=("窗口A",)),
    threading.Thread(target=serve, args=("窗口B",)),
]

for t in threads:
    t.start()
for t in threads:
    t.join()

两个窗口的计数各自从 0 涨到 3，谁也不会把谁的数字改乱——这种数据天生免锁。

13. GIL：多线程头顶的天花板

终于要直面那个绕不开的问题了：CPython 里有个 GIL（Global Interpreter Lock，全局解释器锁），它规定任一时刻、一个进程里只有一个线程能执行 Python 字节码。

听起来多线程好像白搭？做个实验最直观——纯计算任务，双线程对比单线程连跑两遍：

import threading
import time


def pixel_math() -> int:
    """模拟像素级运算——纯 CPU 计算"""
    total = 0
    for i in range(30_000_000):
        total += i * i
    return total


# 双线程版本
t1 = threading.Thread(target=pixel_math)
t2 = threading.Thread(target=pixel_math)

start = time.time()
t1.start()
t2.start()
t1.join()
t2.join()
print(f"两个线程：{time.time() - start:.2f} 秒")

# 单线程版本
start = time.time()
pixel_math()
pixel_math()
print(f"单线程串行：{time.time() - start:.2f} 秒")

跑完大概率会发现：双线程不但没快，可能还略微慢一点（多了切换开销）。

那 GIL 之下，多线程到底图什么？图"等"。线程遇到 I/O（网络、磁盘）时会释放 GIL，把执行权让给别的线程。所以对 I/O 密集型任务，多线程照样省时间；但对 CPU 密集型任务，Python 多线程用不上多核，纯属白忙。

14. CPU 密集型的正解：多进程

绕开 GIL 的思路很直接：锁的是"一个进程里的线程"，那就多开几个进程。每个进程有独立的解释器、独立的 GIL，各占一个核，互不牵制：

from multiprocessing import Process


def pixel_math() -> int:
    total = 0
    for i in range(30_000_000):
        total += i * i
    return total


if __name__ == "__main__":  # 多进程程序必须加这一行保护
    p1 = Process(target=pixel_math)
    p2 = Process(target=pixel_math)
    p1.start()
    p2.start()
    p1.join()
    p2.join()

API 和 threading.Thread 几乎一模一样，上手零成本。代价是进程更重、启动更慢，进程间通信还得走 IPC，所以别一上来就开一大堆。

15. 死锁：谁也不撒手，谁也走不了

锁用多了会出新毛病。设想办公室只有一台打印机和一台扫描仪，两位同事都要同时用到两台设备，但占用的顺序相反：

import threading
import time

printer = threading.Lock()
scanner = threading.Lock()


def colleague_a():
    with scanner:          # 甲先占扫描仪
        time.sleep(0.1)
        with printer:      # 再等打印机
            print("甲：扫描、打印，一条龙完成")


def colleague_b():
    with printer:          # 乙先占打印机
        time.sleep(0.1)
        with scanner:      # 再等扫描仪——两人僵住了
            print("乙：打印、扫描，一条龙完成")


ta = threading.Thread(target=colleague_a)
tb = threading.Thread(target=colleague_b)
ta.start()
tb.start()
ta.join()
tb.join()
# 程序永远停在这里：
# 甲攥着扫描仪等打印机，乙攥着打印机等扫描仪，谁也等不来

这就是死锁：两个线程各握一把锁，又都在等对方手里的那把。破局的原则简单到一句话——所有线程按同样的顺序拿锁：

import threading
import time

printer = threading.Lock()
scanner = threading.Lock()


def colleague_a():
    with scanner:          # 统一先拿扫描仪
        time.sleep(0.1)
        with printer:
            print("甲：扫描、打印，一条龙完成")


def colleague_b():
    with scanner:          # 乙也先拿扫描仪——顺序一致，僵局不再
        time.sleep(0.1)
        with printer:
            print("乙：打印、扫描，一条龙完成")


ta = threading.Thread(target=colleague_a)
tb = threading.Thread(target=colleague_b)
ta.start()
tb.start()
ta.join()
tb.join()
print("两件事都办完了")

顺序统一后，后到的人会在第一把锁处排队，而不是各占一半互相耗着。

16. 选型速查表

任务长什么样	顺手的工具
CPU 密集（计算、编解码、图像处理）	`multiprocessing` / `ProcessPoolExecutor`
同步 I/O 密集（阻塞的文件操作、数据库驱动）	`threading` / `ThreadPoolExecutor`
异步 I/O 密集（爬虫、高并发 Web 服务）	`asyncio` 协程
协程里必须调用阻塞库	用 `asyncio.to_thread` / `run_in_executor` 包一层

到这里，操作系统层面的并发工具就齐了。但你多半已经注意到：速查表里"异步"一栏反复出现的 asyncio、协程、事件循环，一个都还没拆开——尤其那个贯穿始终的事件循环，它凭什么能用一个线程扛住成千上万的并发任务？下一章，我们就掀开这台发动机的盖子。

下一章：19 · 事件循环