前言multiprocessing里的类不少但真正写业务代码时你最常打交道的其实只有一个——进程池Pool。原因是手动Process起一批进程、喂任务、收结果、控制并发度这四件事写起来啰嗦又容易出 bugPool把「固定数量的工作进程 任务队列 结果收集」封装好了你只要把任务函数和一批数据交给它。本文标题里的空格和下划线都照原样保留因为重点是讲清一个容易混的用法边界Pool(n)里那个n是同时活着的工作进程数不是「任务总数」也不是「CPU 核数」的固定倍数。另一个常见误解是以为map和imap只是「同步/异步」的差别——其实真正的区别是惰性与内存map会先把所有结果收集成一个列表再返回imap返回迭代器、边算边给。本文围绕Pool展开怎么创建、map/imap/imap_unordered怎么选、apply/apply_async与.get()的配合、close()与join()为什么要成对、以及with Pool() as p:这个上下文管理器到底做了什么。示例基于 Python 3.8 及以上。补一句旧版本对照Python 2.7 已于2020 年 1 月 1 日停止维护。如果是从老项目迁过来的代码注意两处——Pool本身在 Python 2 里也有、map一样返回列表不必改写但老代码里紧跟着的print result是 Python 2 的语句形式迁到 Python 3 要改成print(result)。一、创建池与关闭池# 适用于 Python 3.8import multiprocessing as mpdef price(order):工作进程里执行算一个订单的金额。order 是 (数量, 单价)。qty, unit orderreturn qty * unitif __name__ __main__:orders [(3, 10.0), (1, 25.5), (7, 4.0), (2, 99.0)]pool mp.Pool(processes2) # 常驻 2 个工作进程amounts pool.map(price, orders)pool.close() # 不再接收新任务pool.join() # 等工作进程退出print(amounts)Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])的常用参数参数作用备注processes工作进程数量3.13 起默认用os.process_cpu_count()之前是os.cpu_count()initializer每个工作进程启动时调用的函数用来做「每个进程只做一次的初始化」initargs传给initializer的参数元组与前者配套maxtasksperchild一个进程最多处理多少任务后重启默认None即一直活着close()与join()为什么必须成对官方文档写得很清楚close()是「不再向池里提交新任务等已提交的任务做完后工作进程退出」join()是「等待工作进程退出」并明确要求必须先调用close()或terminate()才能用join()。也就是说close()负责「停止收单」join()负责「等人走完」两者分工不同。还有一条来自文档的警告不能忽略池对象有内部资源需要正确管理要像上下文管理器一样用或手动调close()/terminate()否则可能在进程收尾时挂住。文档还特别说明不要把清理工作寄托在垃圾回收上——CPython 不保证池的终结器一定会被调用。二、map、imap 与 imap_unordered这三个方法的差别一句话概括map一次性返回列表imap返回惰性迭代器imap_unordered返回的迭代器不保证顺序。# 适用于 Python 3.8import multiprocessing as mpdef load_line(text):模拟处理一行文本并返回它的长度。return len(text)if __name__ __main__:lines [fline-{i} * (i 1) for i in range(6)]with mp.Pool(3) as pool:# map阻塞到全部算完返回列表顺序与输入一致lengths pool.map(load_line, lines)print(map:, lengths)# imap返回迭代器边算边取顺序与输入一致for value in pool.imap(load_line, lines, chunksize2):print(imap:, value)# imap_unordered谁先算完谁先出顺序不保证for value in pool.imap_unordered(load_line, lines, chunksize2):print(无序:, value)各方法的关键点map(func, iterable[, chunksize])它是内置map的并行版只接收一个可迭代参数要处理多参数就用starmap。它会等全部完成再返回一个列表因此超长可迭代对象会占用较多内存——文档专门提醒了这一点并建议这种情况改用imap或imap_unordered并显式给chunksize。imap返回迭代器逐个产出结果顺序与输入顺序一致。文档说明它支持chunksize且当chunksize1时迭代器的next()可以带超时参数超时抛multiprocessing.TimeoutError。imap_unordered和imap一样是迭代器唯一区别是结果顺序视为任意。文档补充了一句值得玩味的说明只有当池里只有一个工作进程时顺序才是「正确」的。它的好处是「先算完的先返回」不必为了保持顺序而排队等待。chunksize是「一次派给某个工作进程多少个任务」。默认值较小任务很多时适当调大能减少调度往返但调得太大又会让负载不均——最后一批任务可能只落在一个进程上别人闲着。三、starmap多参数任务map只能喂一个可迭代参数任务函数要多个参数时用starmap它把每个元素当作「一个参数序列」拆开。# 适用于 Python 3.8import multiprocessing as mpdef discounted(qty, unit, rate):工作进程里执行数量 × 单价 × 折扣率。return qty * unit * rateif __name__ __main__:jobs [(3, 10.0, 0.9), (1, 25.5, 1.0), (7, 4.0, 0.8)]with mp.Pool(2) as pool:# 每个元素是一个元组被拆成三个位置参数totals pool.starmap(discounted, jobs)print(totals)文档对starmap的说明是它像map但可迭代对象里的每个元素本身应该是可迭代的会被拆包成参数——所以[(1, 2), (3, 4)]的结果是[func(1, 2), func(3, 4)]。starmap是 Python 3.3 起提供的。四、apply 与 apply_asyncapply与apply_async是「单任务」接口与map系列针对「一批任务」不同方法是否阻塞返回值重要限制apply(func[, args[, kwds]])阻塞到算完直接结果只在一个工作进程里执行apply_async(func[, args[, kwds[, callback[, error_callback]]]])立即返回AsyncResult对象用.get()取结果apply有一个反直觉的点官方文档明确写了func只会在池中的某一个工作进程里执行。也就是说apply并不会把任务「摊开」并行——它只是「借用池里的一个进程来跑」。真要并行用apply_async一口气提交多个再统一收结果。# 适用于 Python 3.8import multiprocessing as mpdef slow_double(x):return x * 2if __name__ __main__:with mp.Pool(4) as pool:# 一口气提交 5 个任务它们可能落在不同工作进程上handles [pool.apply_async(slow_double, (i,)) for i in range(5)]# AsyncResult.get([timeout]) 取结果超时抛 multiprocessing.TimeoutErrorresults [h.get(timeout5) for h in handles]print(results)apply_async返回的是AsyncResult它提供get([timeout])取结果若远端抛了异常会在这里重新抛出、wait([timeout])只等就绪、ready()与successful()。它还能传两个回调callback在成功时收到结果error_callback在失败时收到异常实例。文档提醒回调要「立即完成」否则处理结果的线程会被阻塞住。五、上下文管理器写法Pool自 Python 3.3 起支持上下文管理协议。但要看清__exit__做了什么文档写明__enter__()返回池对象而__exit__()调用的是terminate()——不是close()join()。# 适用于 Python 3.8import multiprocessing as mpdef work(x):return x 1if __name__ __main__:# 退出 with 块时会终止池块内提交的任务若已完成就没事with mp.Pool(3) as pool:values pool.map(work, range(6))print(values) # 在块内取结果来得及# 出了这个块池已经被终止由此得出一条实用规则用with Pool()时结果要在块内取完。因为map本身就是阻塞的块内pool.map(...)能拿到完整结果但如果你在块内只提交了apply_async还没来得及.get()出了块池就被terminate()任务可能没跑完就被打断。六、Pool 与 ProcessPoolExecutor 怎么选concurrent.futures.ProcessPoolExecutor在Pool之上提供了一层更统一的抽象把「进程池」和「线程池」放到同一套接口submit/map/as_completed下。维度multiprocessing.PoolProcessPoolExecutor结果抽象AsyncResultFuture等待全部close()join()with块退出或shutdown(waitTrue)边完成边处理imap_unorderedas_completed(futures)超时取结果AsyncResult.get(timeout)Future.result(timeout)与线程池统一接口否是ThreadPoolExecutor同构长驻 worker 数上限无特别说明Windows 上max_workers必须不超过 61ProcessPoolExecutor(max_workersNone, mp_contextNone, initializerNone, initargs(), max_tasks_per_childNone)的默认max_workers是os.process_cpu_count()3.13 起此前是os.cpu_count()。文档列了几条硬性限制只有可 pickle 的对象才能被提交和返回__main__模块必须能被工作子进程导入因此它在交互式解释器里不工作在提交给执行器的可调用对象里再调用Executor或Future的方法会导致死锁lambda和在 REPL 里定义的函数都不要指望能跑。它之所以能绕开 GIL是因为底层就是multiprocessing。文档里那句话说得很到位ProcessPoolExecutor用multiprocessing模块因此能绕开全局解释器锁但也意味着只有可 pickle 的对象才能被执行和返回——收益与约束是一体两面。常见坑点坑点 1在with Pool()块里只提交异步任务、不等结果就出块。❌with Pool(4) as p: p.apply_async(f, (1,))——__exit__调用的是terminate()任务可能被打断。 ✅ 结果在块内用.get()收齐要「提交后再等」就用close()join()的显式写法。坑点 2忘了close()或join()。❌ 手动创建池后直接结束程序 —— 文档警告可能因内部资源未正确管理而在收尾时挂住。 ✅ 用with Pool()上下文或严格按close()→join()收尾。坑点 3以为apply会并行。❌ 循环里反复调pool.apply(f, ...)以为在并行 —— 每次都是在某一个工作进程里同步跑完才返回。 ✅ 要并行就批量提交apply_async再统一.get()或直接用map/imap。坑点 4依赖imap_unordered的顺序。❌ 用imap_unordered却按下标去对应输入 —— 文档说明它的顺序应视为任意。 ✅ 需要顺序就用map/imap不需要顺序、只要「先算完先处理」时才用无序版。坑点 5给超长可迭代对象用map。❌pool.map(f, 上千万个元素)——map会先把全部结果收成一个列表内存压力大。 ✅ 改用imap/imap_unordered并显式设置chunksize文档正是为此推荐的。坑点 6在子进程里调用池的方法。❌ 把池对象当参数传给另一个进程再map—— 文档明确池的方法只应由创建它的进程调用。 ✅ 池只在创建它的进程里使用子进程只接收数据、返回数据。坑点 7任务函数不可 pickle。❌ 用lambda或定义在函数内部的嵌套函数当任务 —— 子进程无法按名字定位它序列化失败。 ✅ 任务函数定义在模块顶层参数也必须是可 pickle 的文件对象、生成器不行。坑点 8以为任务里的异常会被吞掉。❌ 用apply_async提交后从不.get()—— 远端抛的异常要等到.get()才被重新抛出不取就永远发现不了。 ✅ 用map/starmap让异常当场冒泡或对每个句柄都调.get()apply_async还可配error_callback。总结你的场景用哪个关键点一批数据、同样的处理map阻塞返回列表顺序一致注意内存一批数据、边算边处理imap惰性迭代器顺序一致一批数据、先算完先要imap_unordered顺序任意别依赖每个任务多个参数starmap元素本身是可迭代的会被拆包单个任务apply/apply_asyncapply阻塞且只用一个进程异步用.get()收尾close()join()join前必须先close或terminate简洁写法with Pool() as p:退出时terminate()结果要在块内取完想要统一接口ProcessPoolExecutorFutureas_completed与线程池同形Pool的核心就三件事用几个进程、任务怎么发、结果怎么收。进程数由processes定任务发放看是「一批」还是「单个」map系列 vsapply系列结果回收看要不要顺序、要不要惰性map/imap/imap_unordered。真正容易出错的不是 API 记不熟而是两个细节收尾用with时__exit__走的是terminate()所以别把没取结果的任务留到块外以及所有跨进程传递的东西都必须可 pickle。把这两条盯住池的用法就基本不会翻车。