multiprocessing进程池实战

可视化演示

展示全概率公式P(A)=ΣP(A|Bi)P(Bi)的计算过程，将样本空间划分为互斥事件Bi，通过加权求和得到事件A的总概率。

进程池基础：组建你的后厨班底

老板决定开一家川菜馆，第一件事就是定班底。他不会每天都重新招人，而是固定雇好 4 位厨师，让岗位一直有人——这就是进程池的“进程复用”思想：创建一批常驻子进程，反复接单，而不是每来一单就现招一个人（那是 Process 裸用，创建销毁成本极高）。

翻译成 Python，multiprocessing.Pool(processes=4) 就是这次招聘。Pool 创建时会 fork 出 4 个子进程，每个子进程都有自己独立的 Python 解释器和内存空间，好比每位厨师有自己的灶台和调料柜。随后你把任务“递进传菜口”，空闲的厨师会自动取单、执行、把结果端到出菜口，全程不需要你干预。

具体来说，Pool 给出两大点单接口：map 把一个函数批量套到整个列表上；apply 传具体参数执行一次。两者都会自动派单。要注意的是，子进程是主进程的“复制品”：模块里的全局变量在 fork 时被复制了一份，大家在各自的副本上操作，谁也看不见谁的改动——就像每位厨师手里各有一张复印的菜单，改自己那份不影响别人。

所以后厨创业有个基本觉悟：订单太少时，单锅快炒比开团队更划算；只有订单源源不断，团队成本才摊得薄。这正是进程池的适用边界：任务量要够大、够密集，否则你付的“培训费”比省下的时间还多。

map 批量映射：一百份同样的宫保鸡丁

中午来了一个百人旅行团，每人点同一道宫保鸡丁。老板做了一件聪明事：不一份份地跟厨师交代，而是把 100 张单子合成一摞递进传菜口，附一句“全部做成宫保鸡丁”。这就是 map 的运作方式——把一个函数应用到可迭代对象的每一个元素上。

在代码里，pool.map（函数，列表） 接受两样东西：一个处理函数，一串待处理的数据。各个子进程从内部队列取一个（或一块）数据，执行函数，把结果放回队列。等所有人干完，pool 按输入顺序把结果排好还给你——第 i 个返回值永远对应第 i 个输入。就像出菜口按点单号摆盘，厨师炒菜的顺序可能千奇百怪，端出来的顺序绝对不乱。

如果函数需要多个参数，比如“统计区间 [a,b) 里的素数个数”，就用 pool.starmap（函数，[(a1,b1）,(a2,b2),……])，它会把每个元组拆开当参数。还有一个小旋钮 chunksize：默认可能一个任务一块，也可以设置成“一次端 10 盘”，减少派单次数，适合单任务极短、任务量极大的场景。

map 就像“标准化菜单”：所有菜品规格统一、流程一致时，批量派单效率最高。如果你的任务各不相同、参数五花八门，那 map 的整齐队形反而派不上用场。

apply 灵活调用：单点定制菜

不是所有客人都点套餐。有人单独点一碗“不要香菜不要葱的牛肉面”，这种单笔定制订单，适合用 apply。老板把这张特殊单子交给后厨，然后站在窗口前等——这段等待就是阻塞：apply(func, args) 提交后，主进程会一直等到结果出锅才继续干别的事。

如果老板不想干等，可以用 apply_async(func, args) 下单，拿到一个“取餐号”（AsyncResult 对象）。你可以先去招呼别的客人，等有空了再凭号取餐：调用 .get() 会阻塞到结果真正就绪；或者干脆留下电话号码——在参数里传 callback 函数，结果一炒好，子进程立刻通过回调通知主进程，主进程不用盯着窗口。

多个 apply_async 同时发出去时，池子会自动把单子分给空闲厨师，实现“多发单、齐开工”。这和 apply 逐个排队等结果形成鲜明对比：同步是“一手交钱一手交货”，异步是“先下单、做完了叫我”。

一句话总结：apply 是“点菜—等菜—吃菜”的餐馆体验，apply_async 是“点外卖—干别的—被叫号”的现代生活。异步的关键收益，是把你等菜的时间也利用起来。

进程间通信：传菜窗口与专用滑道

后厨和前台之间怎么配合？最土的办法是前台亲自跑到后厨喊一嗓子，但人多时必然撞车。餐厅标配是传菜窗口：两边都能往里放、往外取，系统保证不会两份菜互相挤洒——这就是 Queue（队列）。Queue 是进程安全的 FIFO，q.put() 放任务、q.get() 取任务，底层用管道加锁实现，天然支持多生产者、多消费者。

另一种场景：后厨和某个特定档口要高频传菜，专门拉一条滑道，两点直连，速度更快——这就是 Pipe（管道）。Pipe() 返回连接两端的 conn1、conn2，可以单向也可双向传。但它只有两个端点，多个进程抢一根管子时，必须自己加锁，否则数据就穿乱了。

最经典的组合是“生产者—消费者”模型：一个（或几个）进程专职产数据，几个进程从 Queue 里取数据加工，队列充当缓冲，消化掉双方的速度差。速度不匹配会怎样？看下面的公式——整个系统的吞吐，不取决于最快的一环，而取决于最慢的一环。

所以调优时别只盯着生产者死命加速：如果消费者是瓶颈，把菜做得再快也只是让窗口堆成山。找到最慢的一环，才是传菜优化的正确姿势。

共享内存方案：一块共用的白板

传菜窗口每次传“大件货”都很费劲——几 GB 的数组要是走 Queue，得先格式化（pickle 序列化）、搬运到窗口、再拆开还原，来回好几趟。后厨的办法是：在墙上挂一块大白板，所有厨师抬眼就能看到今日材料存量，不用人手一份复印。这就是共享内存。

multiprocessing 提供两个基础构件：Value（类型码，初始值） 共享一个标量（一个数字、一个布尔），Array（类型码，初始数组） 共享一块定长数组。它们存放在所有进程都能直接访问的同一块内存里，读写走的是内存地址，而不是序列化拷贝。代价是“规矩”：任何“读—改—写”操作都要用 Lock 包起来，不然两个厨师同时往白板上加数字，数据就写花了——这叫数据竞争。

适用边界要拎清：共享内存适合数值型、定长的数据（计数器、像素数组、特征向量）；复杂的嵌套对象（字典套列表）不适合，暴力塞进 Array 只会让你怀疑人生，这时候还是老老实实走 Queue。

白板虽然省纸，但所有人都得守规矩：先举手（拿 Lock），再改写，最后放下手。丢掉锁的共享内存，比复制方案更容易把人坑进数据混乱的泥潭。

进程数调优：雇几位厨师最划算

后厨面积有限，灶台数等于 CPU 核心数。你雇 4 位厨师，每个占一个灶，刚刚好；雇 16 位，灶不够用，大家挤来挤去，光让位子就耗掉一半精力——这就是进程数超过核心数后的“上下文切换”惩罚。反过来雇 1 位，订单又积压。多少合适？看任务类型。

如果任务全是纯计算（素数、矩阵乘），一位厨师从头到尾占着一个灶，进程数取核心数最划算。如果任务经常等外部（读文件、等网络响应），厨师在等的时候灶台空着，就可以多雇几个“轮流等”——进程数取核心数乘上一个系数，系数由“等待时间 / 计算时间”决定。

但理论归理论，每台机器、每类任务的开销都不一样。最务实的做法是实测：进程数取 1、2、4、8、16 各跑一轮，画一条“进程数—耗时”曲线，找到谷底。另外别忘了算内存账：每个 Python 进程自带几十 MB 常驻内存，进程开多了，内存账单先爆。

开店的智慧是“先试营业再定编制”。别迷信某个固定值，让数据告诉你答案——跑一组进程数扫描，耗时曲线的最低点，就是你的最优班底规模。

踩坑避雷：后厨翻车现场

后厨经营久了，总有几个经典翻车瞬间。第一次翻车：主厨把菜谱写得“只可意会不可言传”——你在代码里写了个 lambda 塞给 Pool，子进程一脸懵：这菜谱没法抄送（pickle 序列化失败）。因为进程间传函数靠 pickle 按名字导入，lambda 没有名字，传不过去。解法是把逻辑写成模块级别的 def 函数。

第二次翻车：厨师下班后赖在休息室不走。进程池里的子进程不会因为“没活干”就自动消失，你不 close() 关掉接单窗口、不 join() 等人走完，它们就成了僵尸进程，占着进程表不撒手。最省心的是用 with Pool() as pool 上下文管理器，块结束时自动收尾。

第三次翻车：全员僵在原地互相等。经典死锁是“队列没排空就 join”——子进程还在等队列里的任务，主进程却开始等子进程结束，两边干瞪眼。用 with 语句、先 close 再 join、确保队列任务能被消费完，才能避免这种互相等待。最后提醒：进程是有内存成本的，开池前算算内存账单。

打烊前要做三件事：清点厨师（join 回收）、关火关气（close 关池）、核对账单（留意内存）。后厨管理做扎实，流水线才跑得长久。

AI 应用场景

Python 演示

import timefrom multiprocessing import Pool, Queue, Process, cpu_count  def is_prime(n):    """朴素素数判断, CPU 密集型小任务。"""    if n < 2:        return False    for i in range(2, int(n ** 0.5) + 1):        if n % i == 0:            return False    return True  def primes_in_range(start, end):    """统计半开区间 [start, end) 内素数个数。"""    return sum(1 for x in range(start, end) if is_prime(x))  def worker_sum(q_in, q_out):    """Queue 消费者: 从 q_in 收数字并累加, 结果放进 q_out。"""    total = 0    while True:        num = q_in.get()        if num is None:            q_out.put(total)            break        total += num  def main():    parts = min(cpu_count() or 2, 8)    limit = 400000    step = limit // parts    jobs = [(i * step, (i + 1) * step) for i in range(parts)]     t0 = time.time()    with Pool(parts) as pool:        counts = pool.starmap(primes_in_range, jobs)    print("进程池统计素数:", sum(counts), "耗时 %.3f s" % (time.time() - t0))     t0 = time.time()    serial = sum(primes_in_range(s, e) for s, e in jobs)    print("串行统计素数:", serial, "耗时 %.3f s" % (time.time() - t0))     q_in = Queue()    q_out = Queue()    procs = [Process(target=worker_sum, args=(q_in, q_out))             for _ in range(parts)]    for p in procs:        p.start()    for i in range(1, parts + 1):        q_in.put(i)    for _ in procs:        q_in.put(None)    result = 0    for _ in procs:        result += q_out.get()    for p in procs:        p.join()    print("Queue 通信求和:", result)  if __name__ == "__main__":    main()

1. with Pool(parts) as pool:创建进程池并自动清理，等价于“招厨师且打烊自动收尾”，避免僵尸进程。
2. pool.starmap(primes_in_range, jobs)批量派单：把每个 (start, end) 元组拆开传给函数，像把 100 张宫保鸡丁单子一次塞进传菜口。
3. Process(target=worker_sum, args=(q_in, q_out))手动创建子进程并接上传菜窗口，演示 Pool 之外的底层玩法。
4. q_in.put(None)放入“结束信号”，让每个 worker 在队列里读到 None 后主动下班，这是优雅停止消费者的惯用技巧。
5. p.join()等子进程回收，防止厨师赖在休息室变成僵尸进程。
6. if __name__ == "__main__":多进程入口保护：Windows 上 fork 子进程会重新导入本模块，没有这层保护会无限递归创建进程。

常见误区

面试题