Python · 多进程 multiprocessing

场景 1 · 12 万张图片压缩, 4 核跑到 3.8x

图片解码/重编码是纯 CPU 活, 线程被 GIL 摁住。进程池一开, 每个核一个解释器:

from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from PIL import Image

def compress(path: str) -> int:
    img = Image.open(path)                  # 解码+重编码全程吃 CPU, GIL 主场
    out = Path(path).with_suffix(".webp")
    img.save(out, "WEBP", quality=80)
    return Path(path).stat().st_size - out.stat().st_size

if __name__ == "__main__":                # spawn 模式的入场券, 不写必炸
    with ProcessPoolExecutor(max_workers=4) as ex:  # 4 核 = 4 个独立 GIL
        saved = sum(ex.map(compress, paths, chunksize=16))
    print(f"freed {saved/1e9:.1f}GB")
# 实测: 单进程 38min → 4 进程 10min (3.8x); 线程版只有 1.2x

场景 2 · 30GB 文件分块打分, imap_unordered 流式回收

map 要等全量结果才返回, 大文件直接把内存顶爆。imap_unordered 谁先算完谁先回来:

from multiprocessing import Pool

def read_blocks(path: str, size: int):        # 生成器: 逐块读, 不物化整个文件
    with open(path, "rb") as f:
        while blk := f.read(size):
            yield blk

def score(item: tuple[int, bytes]) -> tuple[int, float]:
    i, blk = item
    return i, heavy_model(blk)               # 纯 CPU, 无共享状态, 随便并行

with Pool(4) as pool:
    jobs = enumerate(read_blocks("big.bin", 1 << 20))   # 1MB 一块
    for i, s in pool.imap_unordered(score, jobs, chunksize=8):
        sink.write(f"{i}\t{s}\n")      # 流式落盘, 全程只驻留几个块

内存恒定在几十 MB, 而且乱序回收意味着慢块不阻塞快块的结果落盘。

场景 3 · map 的 chunksize 调优, 小任务合并过桥

任务本身只有 0.2ms, 但每次分发都要 pickle 参数 + 回传结果, IPC 是每任务固定税:

import math
from multiprocessing import Pool

n = 1_000_000
# chunksize=1: 一百万次"发送-回传"往返, IPC 占一半以上时间
with Pool(4) as pool:
    r1 = pool.map(normalize, range(n), chunksize=1)      # 21.4s
with Pool(4) as pool:
    r2 = pool.map(normalize, range(n), chunksize=1000)   # 5.4s, 同样的计算

# 起步值: math.ceil(n / (workers * 4)), 再压测微调
suggested = math.ceil(n / (4 * 4))
# 注意: chunksize 过大会让先领大块的 worker 忙死、别人饿死, 损失负载均衡

场景 4 · SharedMemory 给子进程传 32MB numpy 数组, 零拷贝

用 Queue 传大数组等于 pickle 序列化一份 + 对端再反序列化一份。共享内存只传一个名字:

from multiprocessing import Pool, shared_memory
import numpy as np

arr = np.random.rand(2000, 2000)                 # ~32MB
shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
shared = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
shared[:] = arr                 # 拷一次进共享区, 之后所有进程直接读

def worker(args: tuple[str, tuple]) -> float:
    name, shape = args
    shm = shared_memory.SharedMemory(name=name)     # 按名字 attach, 不过 pickle
    view = np.ndarray(shape, dtype=np.float64, buffer=shm.buf)
    r = view.mean()
    shm.close()                 # 只解除本进程映射, 不回收区域
    return float(r)

# 传给 pool 的只有 shm.name 字符串 + shape, 不是 32MB 数组
shm.close(); shm.unlink()       # 最后一个使用者负责回收, 防 /dev/shm 泄漏

同规模数据 Queue 传 10 次 ~1.8s, SharedMemory 方案 attach 后近乎免费。

场景 5 · 修复 spawn 模式下"模块级状态消失"

同一份代码 Linux 好、Windows/macOS 挂: spawn 子进程重新 import 模块, 模块级初始化不会"继承"过来。把初始化收进 initializer:

# Windows/macOS 默认 spawn: 子进程重新 import 本模块
MODEL = None          # fork 下拿到父进程的值; spawn 下就是 None!

def init_worker(path: str) -> None:
    global MODEL
    MODEL = load_model(path)    # 每个 worker 各自加载一次, 只在启动时付成本

def infer(text: str) -> str:
    return MODEL.predict(text)     # 运行期不再碰全局初始化逻辑

if __name__ == "__main__":   # 不写: 子进程 import 时又执行这里 → 无限起进程
    with Pool(4, initializer=init_worker, initargs=("model.bin",)) as pool:
        results = pool.map(infer, texts)

场景 6 · maxtasksperchild 自愈内存缓涨的 worker

批处理跑两小时后 worker RSS 从 2.1GB 爬到 3.5GB, 第三方库泄漏抓不动。定期让 worker "转世":

# worker 每处理 5000 个任务就重生: 泄漏的内存/句柄随旧进程归还 OS
with Pool(8, maxtasksperchild=5000) as pool:
    for meta in pool.imap(process_page, pages, chunksize=32):
        progress.mark(meta)
# 观测: 改完后 worker RSS 平稳在 2.1GB, 再没触发过 OOM kill
# 代价: 重生要重新 import + 跑 initializer
# 5000 是"自愈频率 vs 重启成本"的折中, 按单任务耗时压测取值

场景 7 · Manager().dict() 做跨进程配置热更新

想在不重启 worker 的情况下调限流阈值。Manager 提供跨进程可见的代理 dict:

from multiprocessing import Manager, Pool
import time

CONFIG = None       # 每个 worker 里都是同一个 Manager dict 的代理

def call_api(task: str) -> int:
    rate = CONFIG["rate_limit"]     # 每次读是一次到 Manager 的 RPC
    time.sleep(1 / rate)
    return len(task)

if __name__ == "__main__":
    with Manager() as mgr:
        CONFIG = mgr.dict({"rate_limit": 100})
        with Pool(16) as pool:
            pool.map(call_api, tasks)
# 运维通道执行 CONFIG["rate_limit"] = 200, worker 下次读取即生效
# 低频读没问题; 高频计数请换 Value/SharedMemory, 别走 RPC

场景 8 · worker 各自建 DB 连接, 别继承父进程的

fork 出的 worker 用父进程连接, 服务端把四个进程当同一会话: 事务串号、command out of sync 连环报错。每个 worker 自己建:

CONN = None

def init_db(dsn: str) -> None:
    global CONN
    CONN = psycopg.connect(dsn)     # 各自建连接: 会话独立, 锁也不共享

def query(sql: str) -> list:
    return CONN.execute(sql).fetchall()

if __name__ == "__main__":
    # initializer 在每个 worker 启动时跑一次, 连接随 worker 生死
    with Pool(4, initializer=init_db, initargs=(DSN,)) as pool:
        rows = pool.map(query, sqls)
# 配 maxtasksperchild 时注意: worker 重生会重连一次, 池要给连接数留余量

场景 9 · apply_async + 回调, 异步任务队列雏形

异步提交最大的坑是异常静默: worker 里抛的错只存在于 result 对象里, 不 get 就蒸发。回调 + 结果对象双保险:

from multiprocessing import Pool

def job(payload: dict) -> dict:
    try:
        return {"ok": True, "data": render(payload)}
    except Exception as e:       # worker 内先兜住, 装进结果回传
        return {"ok": False, "error": repr(e)}

def on_done(res): cache.set(res["id"], res)
def on_fail(exc): alert.fire(f"pool job crashed: {exc!r}")

res = pool.apply_async(job, (payload,),
                       callback=on_done, error_callback=on_fail)
res.get(timeout=30)         # 需要同步语义时再等; 不 get 也别忘 error_callback

场景 10 · asyncio 服务里 CPU 任务不堵事件循环

Web 服务里直接调指纹/加密等 CPU 函数, 事件循环一卡所有并发请求陪葬。扔进进程池:

import asyncio
from concurrent.futures import ProcessPoolExecutor

async def handler(req):
    # 直接 fingerprint(req.body) 会堵死事件循环, 全部请求一起卡
    loop = asyncio.get_running_loop()
    digest = await loop.run_in_executor(POOL, fingerprint, req.body)
    return {"digest": digest}

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as POOL:
        asyncio.run(serve(handler))
# 复用同一个池: 每请求 fork 一个进程的话, 启动开销吃光并行收益

P99 从 3.4s 回到 180ms — 事件循环负责 IO 并发, 进程池负责 CPU 满载, 各干各的。