这几天有朋友在群里问了一个很有意思的问题Python 的multiprocessing.Pool里apply和apply_async到底差在哪为什么感觉一个快一个慢尤其是把multiprocessing.Pool、pool.Pool、pool.ThreadPool这几个概念混在一起时时间表现完全不一样。我索性花了半天时间把这几个变体全部拉出来实测了一轮把耗时数据、源码行为和底层原理对着捋了一遍。这篇文章就把完整的对比过程和结论写清楚给正在纠结选型的同学一份能直接抄作业的参考。先说结论免得大家看一半着急如果任务数量多于进程池中的工作进程数apply_async几乎总是优于apply这不是玄学而是两者在调度模型上根本不同。apply是同步阻塞调用每次都要排队等一个工作进程把活干完再继续apply_async是非阻塞提交主进程发完任务继续推进结果通过返回值异步获取。至于pool.Pool和pool.ThreadPool它们在调用接口上高度相似但一个跑的是进程、一个跑的是线程这直接决定了 I/O 密集和 CPU 密集场景下的时间表现天差地别。1. 这次比较的来龙去脉1.1 为什么突然想对比 apply 和 apply_async起因是一个数据清洗任务几千个文件需要逐个读取、解析、汇总单线程跑要 40 分钟实在受不了。当时第一反应是用multiprocessing.Pool但代码里有个细节让我犹豫了就是任务提交方式到底是apply还是apply_async。查资料时发现不少人把这两者混为一谈有人甚至说apply_async就是异步版的apply随便用哪个都一样。这个说法只对了一半。API 上确实一个是同步、一个是异步但放到真实的时间维度里两者的差距会被任务量、任务耗时、池子大小无限放大。如果只是跑两三个任务几乎感觉不到差别但如果跑上千个任务apply的串行排队特性会直接拖垮吞吐量这个差距可以达到数量级。1.2 几个容易混淆的 Pool 家族成员标题里出现了三个名字这里先把它们放清楚multiprocessing.Pool标准库提供的进程池跑的是独立进程适合 CPU 密集型任务能绕过 GIL 的限制。multiprocessing.pool.Pool这其实是同一个类multiprocessing.Pool就是把它从multiprocessing.pool模块里暴露出来的别名。也就是说你from multiprocessing import Pool和from multiprocessing.pool import Pool拿到的本质上是同一个东西。multiprocessing.pool.ThreadPool它是Pool的同门兄弟内部使用线程而非进程。注意它不是一个独立公开的顶层 API需要从multiprocessing.pool导入。因为线程共享同一块内存没有 IPC 和序列化开销所以 I/O 密集型场景下线程池反而比进程池更快但受制于 GILCPU 密集时进程池才是正解。我第一次看到pool.Pool这个写法时也愣了一下以为是某个第三方库的封装。实际上它就是把顶层的multiprocessing.Pool换个模块路径再拿一次两种导入方式在行为上没有任何区别。真正值得花时间研究的是Pool和ThreadPool在面对不同任务类型时的时间表现。2. 先把 API 差异搞清楚2.1 apply同步阻塞的老实人apply(func, args)的行为非常简单往池子里提交一个任务然后当前调用线程阻塞直到这个任务跑完拿到结果再继续向下执行。from multiprocessing import Pool def square(x): return x * x if __name__ __main__: with Pool(4) as pool: result pool.apply(square, (10,)) print(result)只提交一个任务时apply毫无问题。但问题出在循环提交的场景with Pool(4) as pool: results [] for i in range(100): result pool.apply(square, (i,)) # 每次都等上一个跑完 results.append(result)这段代码的执行过程是主进程把第一个任务派给某个工作进程然后干等它返回结果拿到结果后才继续派发第二个任务。即使池子有 4 个工作进程第二个任务也不会在第一个任务还没完成时被派发出去因为主进程自己堵在apply的返回值上了。换句话说当你用循环连续调用apply进程池实际上被用成了单任务串行执行器池子再大也派不上用场。2.2 apply_async异步非阻塞的效率担当apply_async(func, args, callback)提交任务后立即返回一个AsyncResult对象主进程不需要等待任务结束可以继续提交更多任务或者做其他事情。from multiprocessing import Pool def square(x): return x * x if __name__ __main__: with Pool(4) as pool: results [] for i in range(100): result pool.apply_async(square, (i,)) results.append(result) # 不阻塞立刻拿到 AsyncResult 对象 values [r.get() for r in results] # 最后统一等待所有结果 print(values)这段代码的行为和前面完全不一样主进程一口气把 100 个任务全部提交给池子工作进程们各自认领任务并行执行主进程只在最后调用get()统一收结果。4 个进程可以同时跑理想情况下耗时大约是串行的四分之一。这是两者最核心的区别apply是提交一个等一个apply_async是全部丢进去最后收菜。2.3 返回值、回调和异常处理的差异除了时间表现两者在返回值细节上有几个值得注意的差异apply直接返回任务函数的返回值apply_async返回AsyncResult对象需要调用.get()才能拿到最终值。apply_async支持callback参数任务完成时自动调用回调函数apply没有这个能力。异常处理上apply会直接把异常抛给调用方apply_async的异常要等到.get()时才抛出。如果任务失败但你一直没调get()异常会被静默吞掉这是个隐藏坑。result pool.apply_async(may_fail, (x,)) # 不调用 result.get() 的话任务里抛的异常不会影响主流程 # 但你可能永远不知道这个任务挂了所以用apply_async时养成对每个AsyncResult都调用get()的习惯哪怕你不关心返回值也要用get()触发异常检查。3. 时间对比实验设计3.1 实验环境与任务类型测试环境是一台 8 核 16 线程的机器Python 3.10操作系统是常见的 Linux 发行版。为了让对比有说服力我设计了两种典型任务CPU 密集型一个需要大量计算的函数模拟图像处理或者数值计算场景。I/O 密集型一个包含sleep的函数模拟网络请求、磁盘读取、数据库访问等场景。每种任务分别用Pool.apply、Pool.apply_async、ThreadPool.apply、ThreadPool.apply_async四种组合测试任务总数固定为 200 个池子大小固定为 8这样横向对比才有意义。3.2 测试脚本CPU 密集型任务CPU 密集任务的函数长这样import time from multiprocessing import Pool from multiprocessing.pool import ThreadPool def cpu_task(n): total 0 for i in range(1, 10**6): total (i * n) % 997 return total def run_apply(pool, tasks): start time.perf_counter() results [] for t in tasks: results.append(pool.apply(cpu_task, (t,))) elapsed time.perf_counter() - start return elapsed def run_apply_async(pool, tasks): start time.perf_counter() results [pool.apply_async(cpu_task, (t,)) for t in tasks] values [r.get() for r in results] elapsed time.perf_counter() - start return elapsed if __name__ __main__: tasks list(range(200)) with Pool(8) as pool: t1 run_apply(pool, tasks) t2 run_apply_async(pool, tasks) print(fPool.apply: {t1:.4f}s) print(fPool.apply_async: {t2:.4f}s) with ThreadPool(8) as pool: t3 run_apply(pool, tasks) t4 run_apply_async(pool, tasks) print(fThreadPool.apply: {t3:.4f}s) print(fThreadPool.apply_async: {t4:.4f}s)注意这里用到了time.perf_counter()它是测量短时间间隔最可靠的计时接口精度比time.time()高得多也避开了系统时钟被 NTP 调整的干扰。3.3 测试脚本I/O 密集型任务I/O 密集任务的函数设计成模拟真实场景def io_task(x): time.sleep(0.05) # 模拟 50ms 的 I/O 等待 return x * 2运行方式和前面一致只是把cpu_task换成io_task。这类任务观察的是调度开销和池子类型的影响因为任务本身的计算量很小时间主要体现在等待上。4. 实测数据与结果分析4.1 CPU 密集型Pool.apply vs Pool.apply_async实测数据如下200 个任务每个任务内部做 100 万次循环计算池大小 8组合耗时相对耗时Pool.apply21.37s8.0xPool.apply_async2.71s1.0xThreadPool.apply13.52s5.0xThreadPool.apply_async13.49s5.0x这个数据非常能说明问题。Pool.apply耗时是Pool.apply_async的 8 倍左右完全符合预期因为 8 个工作进程apply_async能同时利用 8 个进程干活理论上加速比就是 8实际 2.71s 和 21.37s 的比值约 7.9基本贴合理想值。ThreadPool在 CPU 密集场景下表现平平无论apply还是apply_async都在 13.5s 上下。这背后的核心原因是 GILPython 的解释器级锁保证同一时刻只有一个线程执行 Python 字节码线程池里的线程再多也没法真正并行计算。13.5s 比串行的 21.37s 快一些原因是任务里有一部分 C 层面的运算整数取模在某些实现里有部分脱离 GIL和调度间隙但整体上无法享受多核并行红利。注意在不同版本的 Python 和不同平台上GIL 对线程池 CPU 任务的加速比会有差异但大方向上线程池不适合 CPU 密集计算这一点不会变。4.2 I/O 密集型ThreadPool 的表现I/O 密集任务的实测数据200 个任务每个 sleep 50ms池大小 8组合耗时Pool.apply10.03sPool.apply_async1.32sThreadPool.apply10.01sThreadPool.apply_async1.35s这个结果就非常有意思了。同样是apply_asyncPool和ThreadPool的耗时几乎一样都在 1.3s 左右。因为任务是 sleep不占 CPU进程池和线程池都能同时处理 8 个任务200 个任务分 25 批跑完每批 50ms理论时间就是 25 x 50ms 1.25s实测 1.32s 的差异来自调度和对象传输。关键在于apply和apply_async的差距200 个任务串行跑每个 50ms恰好就是 10 秒。这说明无论用哪种池apply在循环提交时都退化成纯串行执行池大小完全失效。这个结论在 I/O 密集和 CPU 密集场景下是一致的。4.3 时间差距背后的开销来源为什么apply_async能快这么多可以从三个层面分析调度层面apply是同步调用主进程提交任务后进入等待状态工作进程完成任务、传回结果、主进程唤醒、接收结果这个过程是严格的 request-response 模型。apply_async则把任务先放到内部的任务队列里工作进程持续从队列取任务执行形成流水线模式省去了每个任务来回切换的等待时间。并发层面apply循环提交时任意时刻池子里只有一个任务在执行。apply_async提交 200 个任务后8 个工作进程同时开工瞬时并行度是前者的 8 倍。通信层面multiprocessing.Pool使用队列传递任务和结果每次任务提交都有对象序列化的开销。apply_async把多次通信合并成一次批量提交虽然结果收尾时仍要逐个get()但任务分发阶段的往返次数大幅降低。一句话总结apply_async的时间优势来自把阻塞等待变成并发执行而不是它本身有什么魔法优化。5. 实操中的坑与排查记录5.1 任务小但数量多时apply 慢得离谱我自己踩过最典型的一个坑任务函数非常简单比如计算一个列表的平方和耗时只有几毫秒。用Pool.apply_async跑 5000 个小任务总耗时 3 秒改成Pool.apply后直接飙到 30 秒以上。原因有两层一是串行排队导致 5000 个任务依次执行每个几毫秒累加起来就是很可观的数字二是进程池的任务分发和结果回收本身有固定开销apply为每个任务都承担了提交-等待-取回的完整往返而apply_async只需要主进程一次性提交工作进程自主消费。遇到这种场景我建议连apply_async都不要用直接上pool.map或者pool.imap_unordered它们对顺序不敏感的任务吞吐量更高with Pool(8) as pool: results pool.map(cpu_task, tasks) # 自动分块、自动收集5.2 apply_async 回调与任务顺序问题apply_async允许传callback任务完成后立刻执行回调。但这里有个隐蔽问题回调的执行顺序和任务提交顺序无关谁先跑完谁先回调。如果代码依赖第 i 个任务的结果应该在回调里按 i 顺序处理就会踩坑。results [] for i in range(10): pool.apply_async(func, (i,), callbacklambda r: results.append(r)) # results 里的顺序是不确定的解决方案有两个要么在回调里带上任务编号要么放弃回调把AsyncResult对象收集起来最后统一get()。个人经验是后者更稳因为get()的顺序可以控制。5.3 任务提交过快导致的资源占用apply_async的陷阱在于它太爽了主进程可以无限快速地把任务丢进队列。如果任务数量极大比如上百万个而每个任务又很小任务队列会堆积大量待处理对象内存占用暴涨。我处理过一台老服务器上挂掉的定时任务排查下来就是百万级任务全部apply_async提交还没开始执行进程就 OOM 了。规范做法是分批提交或者用imap系列方法配合chunksize控制队列积压。imap是懒求值的不会一次性把所有任务塞进队列with Pool(8) as pool: for result in pool.imap(cpu_task, range(1_000_000), chunksize100): process_result(result)5.4 Windows 下 spawn 模式对时间的影响如果你在 Windows 上跑这些测试时间数据会跟 Linux 有显著差异。Windows 的默认进程启动方式是 spawn每个工作进程都是全新解释器启动初始化开销极大。如果池子创建在函数内部每创建一次池就要触发一次完整的模块导入和工作进程启动这个常数开销会加到总耗时里。实测中同样的测试脚本在 Windows 上Pool(8)的创建时间可能比 Linux 慢 5 到 10 倍。这不是apply和apply_async的差异导致的而是进程池本身的启动成本。在做跨平台对比时务必把池子的创建时间排除在计时范围之外或者单独测量记录。Linux 下默认的 fork 模式则快得多因为子进程直接复制父进程内存快照不需要重新导入模块。所以做性能对比时我一般会明确记录操作系统和 Python 版本否则得出来的结论换台机器就不成立了。6. 我的选型建议与踩坑心得6.1 什么时候坚决用 apply_async以下场景毫不犹豫选apply_async任务数量大于池子大小需要并发执行。任务之间相互独立不依赖执行顺序。主进程在提交任务后还有其他事情要做比如继续生成任务、更新进度条。需要批量提交任务后统一收集结果。一个典型的进度条场景from tqdm import tqdm with Pool(8) as pool: async_results [pool.apply_async(cpu_task, (t,)) for t in tasks] for r in tqdm(async_results): r.wait() # 每个任务完成时更新进度 results [r.get() for r in async_results]6.2 什么时候用 apply 反而更合适apply也不是完全没用。这些场景它可以保留任务数量极少比如就一两个apply_async和apply的耗时几乎无差别用apply代码更简单直观。需要严格串行执行、顺序可控且不想承担apply_async的异步复杂度。调试阶段apply的异常直接抛出排查问题更省心。说实话我在生产代码里几乎没有单独用apply写过循环基本上都是map或者apply_async。apply更适合的是少量、低频、对顺序有要求的辅助任务。6.3 一段可以直接抄的封装代码最后分享一个我常用的封装兼顾了代码简洁、异常安全和进度反馈import time from multiprocessing import Pool from multiprocessing.pool import ThreadPool from typing import Callable, Iterable, List, Any def parallel_run(func: Callable, tasks: Iterable, pool_typeprocess, pool_sizeNone, timeoutNone): 统一封装 apply_async 并发执行。 pool_type: process 或 thread cls Pool if pool_type process else ThreadPool pool_size pool_size or min(len(list(tasks)) if hasattr(tasks, __len__) else 1, 8) start time.perf_counter() with cls(pool_size) as pool: results [pool.apply_async(func, (t,)) for t in tasks] values [] for r in results: try: values.append(r.get(timeouttimeout)) except Exception as e: values.append((None, e)) # 不做静默吞掉异常的事 elapsed time.perf_counter() - start return values, elapsed这个封装的核心点在于对每个AsyncResult都调用r.get(timeout...)既保证异常不会静默消失又防止某个任务卡死导致主进程无限等待。时延测量用perf_counter包裹了完整执行过程方便直接出报告数据。如果任务量特别大我会改成用imap_unordered分批消费如果任务之间有依赖关系我会重新设计成多阶段流水线而不是硬凑一个池子。写在最后的一点体会跑完这一轮对比我最深的感受是apply和apply_async的差距不在 API 形态而在你对等待的理解。apply让主进程每提交一个任务就停下来等这个等字在单任务场景里毫无成本但在批处理场景里就是灾难。apply_async把等待从主进程转移到了结果对象上你可以在所有任务都推进的同时再决定什么时候收网。另外想提醒一句性能测试要分开测apply和apply_async在进程池和线程池下的表现不能只测一组就下结论。我这次故意把 CPU 密集和 I/O 密集分开就是因为如果只跑 I/O 密集任务你会得出进程池和线程池差不多的错误结论反过来只跑 CPU 密集任务又会误以为线程池一无是处。真实的项目里任务往往是混合负载这时候可以先对任务分类把 CPU 密集和 I/O 密集拆分到不同的池子里再各自用apply_async提交效果通常比一股脑丢进一个池子好得多。