多进程 multiprocessing 基础

上一篇 多线程 threading 基础 讲了多线程受 GIL 限制，不适合 CPU 密集任务。这一篇讲多进程——每个进程一个独立 Python 解释器，各有自己的 GIL，真正并行。适合视频处理、数据分析、机器学习推理等 CPU 密集场景。

一、进程 vs 线程

对比项	线程	进程
共享内存	✅ 天然共享	❌ 各自独立，需要 IPC
创建开销	小（毫秒级）	大（几十毫秒到几百毫秒）
数据传递	直接引用	序列化后再发（pickle）
CPU 并行	❌ 受 GIL 限制	✅ 真正并行
崩溃影响	整个进程挂	只挂当前进程

规则：

• CPU 密集 → 多进程；
• IO 密集 → 多线程或异步。

二、创建进程

from multiprocessing import Process

def worker(name):
    print(f"{name} in pid {os.getpid()}")

if __name__ == "__main__":
    p = Process(target=worker, args=("t1",))
    p.start()
    p.join()

关键：if __name__ == "__main__": 不能省——尤其在 Windows/macOS 上。子进程会重新 import 主脚本，如果没有这个保护，会陷入无限递归创建进程。

三、ProcessPoolExecutor：Pythonic 姿势

跟线程池 API 完全一样：

from concurrent.futures import ProcessPoolExecutor
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True


if __name__ == "__main__":
    numbers = range(10**6, 10**6 + 100)
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(is_prime, numbers))
    print(sum(results))

max_workers 默认是 CPU 核心数。

四、Pool：老一代 API

multiprocessing.Pool 是更老的 API，功能类似：

from multiprocessing import Pool

with Pool(4) as p:
    result = p.map(is_prime, numbers)

新代码优先用 ProcessPoolExecutor——它跟 ThreadPoolExecutor 的接口统一，方便切换。

五、任务函数的要求

传给多进程的函数必须能被 pickle：

• 可以：模块顶层的普通函数、类的方法（类必须是顶层的）；
• 不可以：lambda、内嵌函数、局部类的方法。

# ❌ Pool 里报错
def outer():
    def inner(x):
        return x * 2
    with Pool(4) as p:
        p.map(inner, range(10))

# ✅ 顶层函数
def double(x):
    return x * 2

def outer():
    with Pool(4) as p:
        p.map(double, range(10))

六、共享数据

进程间不共享内存，要传数据用：

队列 Queue（多进程版）

from multiprocessing import Process, Queue

def worker(q):
    q.put(42)

if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    p.join()
    print(q.get())     # 42

Pipe：管道（双向通信）

from multiprocessing import Pipe

parent_conn, child_conn = Pipe()

Manager：共享对象

from multiprocessing import Manager

with Manager() as m:
    shared_dict = m.dict()
    shared_list = m.list()
    # 多个进程可以像本地一样操作它们

Manager 比 Queue 慢——它把所有访问都变成 IPC 调用。只在需要”共享结构化状态”时用。

共享内存

Python 3.8+ 有更高性能的共享内存：

from multiprocessing.shared_memory import SharedMemory

shm = SharedMemory(create=True, size=1024)
shm.buf[0] = 42       # 直接操作字节

一般项目用不到，做高性能数据分析时才需要。

七、数据序列化开销

多进程之间传数据要经过 pickle——这是最大的性能陷阱：

• 传大 numpy 数组：慢；
• 传函数（lambda）：不能 pickle，报错；
• 传数据库连接、文件句柄：不能 pickle。

优化：

• 让子进程”自己去拿数据”（读文件而不是父进程传进来）；
• 数据分片，一次传一小片；
• NumPy 数据用 shared_memory 或 Array/Value；
• 少用 Manager 共享结构。

八、进程池 vs 手动 Process

• 进程池（Pool / ProcessPoolExecutor）：推荐，自动管理生命周期；
• 手动 Process：控制精细，适合特殊场景（长期后台进程）。

九、启动方式：fork vs spawn vs forkserver

• fork（Linux/macOS 默认）：复制父进程所有内存，快，但可能有资源泄漏；
• spawn（Windows 唯一，macOS 3.8+ 默认）：从头启动新进程，慢但干净；
• forkserver：fork 一个”服务器进程”，其他都从它 fork。

import multiprocessing as mp
mp.set_start_method("spawn")     # 显式设置

十、CPU 密集加速实测

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))


nums = [10**7] * 8

# 顺序
start = time.perf_counter()
list(map(heavy, nums))
print(f"顺序：{time.perf_counter() - start:.2f}s")

# 多线程（GIL 阻塞，基本没提升）
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=8) as p:
    list(p.map(heavy, nums))
print(f"多线程：{time.perf_counter() - start:.2f}s")

# 多进程（4 核 CPU 上大概快 3-4 倍）
if __name__ == "__main__":
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=8) as p:
        list(p.map(heavy, nums))
    print(f"多进程：{time.perf_counter() - start:.2f}s")

十一、初始化每个进程

worker 进程启动时想做初始化（连数据库、加载模型）：

def init_worker():
    global model
    model = load_expensive_model()

def worker(x):
    return model.predict(x)

with ProcessPoolExecutor(max_workers=4, initializer=init_worker) as pool:
    ...

十二、任务超时

ProcessPoolExecutor 的 future.result(timeout=...) 会等超时——但不会真的杀掉进程，只是抛 TimeoutError。想强制杀 worker，改用 pool.shutdown(cancel_futures=True)（Python 3.9+），或者用 pebble 第三方库。

十三、joblib：更简洁的并行

数据分析场景推荐 joblib（scikit-learn 用的库）：

from joblib import Parallel, delayed

results = Parallel(n_jobs=-1)(delayed(heavy)(n) for n in nums)

n_jobs=-1 用所有核心。API 比 ProcessPoolExecutor 更紧凑。

十四、一个完整例子：批量图片处理

# imgproc.py
"""对一个目录里的所有 JPG 做缩略图，多进程加速。"""

from PIL import Image
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor, as_completed
import os


def make_thumb(src: Path, dst_dir: Path, size=(300, 300)) -> Path:
    dst = dst_dir / src.name
    with Image.open(src) as img:
        img.thumbnail(size)
        img.save(dst)
    print(f"[pid={os.getpid()}] {src.name} → {dst}")
    return dst


def batch(src_dir: Path, dst_dir: Path, workers: int = None):
    dst_dir.mkdir(parents=True, exist_ok=True)
    imgs = list(src_dir.glob("*.jpg"))

    with ProcessPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(make_thumb, p, dst_dir): p for p in imgs}
        for f in as_completed(futures):
            try:
                f.result()
            except Exception as e:
                print(f"失败：{futures[f]} - {e}")


if __name__ == "__main__":
    batch(Path("photos"), Path("thumbs"))

if __name__ == "__main__": 里调用是 Windows/macOS 上多进程的硬性要求。

十五、常见陷阱

1. 忘 if __name__ == "__main__":：Windows 上会陷入无限创建子进程。
2. 传 lambda 给 Pool：pickle 报错。
3. 进程数远超 CPU 核心：调度开销大，反而变慢。
4. 传大数据：序列化开销比计算还大。让子进程自己读磁盘。
5. 进程池里用 print：多个进程混着输出，看起来乱。用 logging 会好些。
6. 共享 Manager 对象锁死：Manager 内部有锁，大量并发写会锁竞争。改用 shared_memory。
7. 忘 shutdown：进程可能滞留。用 with 或显式 pool.shutdown()。

十六、小结与延伸阅读

• 多进程绕开 GIL，真正并行；
• 优先用 ProcessPoolExecutor；
• 函数必须能 pickle（顶层普通函数）；
• 数据传递有序列化开销，尽量少传；
• 进程数 = CPU 核心数附近；
• 保护主入口 if __name__ == "__main__":；
• CPU 密集用它，IO 密集用 threading/asyncio。

延伸阅读：

• multiprocessing：https://docs.python.org/zh-cn/3/library/multiprocessing.html
• concurrent.futures：https://docs.python.org/zh-cn/3/library/concurrent.futures.html
• joblib：https://joblib.readthedocs.io/

下一篇 异步编程 asyncio 入门 我们讲现代 Python 里最流行的并发方式——协程。