如果一个项目里有大量IO密集型任务很多人第一反应是开线程但遇到CPU密集计算线程在Python里基本是摆设GIL卡得死死的。这时候就得请出multiprocessing多进程让每个进程拿一个完整CPU核心干活。可是真把multiprocess跑起来以后你很快会遇到一个尴尬的问题子进程算完了结果怎么拿回来return在进程函数里写了等于白写进程对象join()之后不会像Thread.join()那样留下什么返回值。本文围绕“多进程返回值获取”这个核心把Python多进程里传结果的几种常见做法全部过一遍Queue、Manager共享对象、Pool.map和apply_async顺带介绍concurrent.futures.ProcessPoolExecutor这种现代化写法。适合刚接触多进程、想在业务代码里落地进程池结果的读者也适合已经被返回值问题坑过一次、想搞明白底层逻辑的人。1. 为什么多进程返回值这么难拿1.1 多进程和线程在“返回值”上的本质区别先拆一个基本概念。线程是共享进程内存空间的线程A里定义一个result []线程B往这个列表里append线程A就能看见。因为大家住在同一套地址空间里对象就是同一个对象。进程不一样。Process(targetmy_func)创建出的子进程是一份完整的内存快照拷贝fork时是写时复制或者全新解释器spawn时子进程里所有变量跟主进程没有半毛钱关系。你在子进程里写了return 42这个42只存在于子进程自己的调用栈里子进程退出栈被销毁主进程什么都没得到。所以“子进程函数里写return主进程想直接接住”这件事从设计上就不成立。用生活类比就是线程是一家人住一间房你喊一声全家都听见进程是几个人各住各的房间你喊破嗓子隔壁也听不见必须用电话线IPC机制才能传话。1.2 return在进程里到底去哪了很多人试过这样写from multiprocessing import Process def compute(x): return x * 2 p Process(targetcompute, args(10,)) p.start() p.join() print(p.exitcode) # 只有退出码没有返回值join()之后你只能拿到exitcode这个值表示子进程是否正常结束0是正常非0是崩了它跟你的业务结果没有关系。子进程函数体里return的值操作系统层面来看根本不存在“返还给父进程”这个机制它只会被当成进程退出状态的一部分丢弃或者你显式用sys.exit(code)也只能传一个0到255的整数作为退出码。所以要把结果拿回来本质思路只有一个子进程把结果写到一个主进程能访问的地方主进程主动去取。这个地方可以是管道、队列、共享内存、文件或者某个进程池对象替你包装好的Future。后面的所有方案其实都是在包装这个思路。1.3 先记住三把钥匙Queue、Manager、Pool接下来的内容会围绕三套API展开multiprocessing.Queue进程间传数据的队列最直接最底层。multiprocessing.Manager在独立进程中维护一组共享对象比如共享的dict、list子进程往里写主进程等它们写完后读取。multiprocessing.Pool进程池封装了任务分发和结果收集map、starmap、apply_async可以让你几乎像写同步代码那样拿返回值。它们不是互相排斥的关系往往在同一个项目里混着用。下面我按实际出镜率从高到低逐个讲。2. 用Queue做结果收集最底层也最可控2.1 一个可以直接跑的Queue示例Queue的原理是多进程在底层维护一个管道或内存队列子进程把对象塞进去主进程从另一端取出来。对调用方来说它就是一个线程安全的FIFO。示例import multiprocessing import time def worker(name, duration, queue): time.sleep(duration) queue.put(f{name} 完成耗时 {duration}s) if __name__ __main__: q multiprocessing.Queue() jobs [ multiprocessing.Process(targetworker, args(任务A, 2, q)), multiprocessing.Process(targetworker, args(任务B, 1, q)), ] for p in jobs: p.start() for p in jobs: p.join() results [] while not q.empty(): results.append(q.get()) print(results)这个例子我建议你亲手敲一遍。它把多进程返回值问题的基本套路完整呈现了子进程不return而是把结果put进Queue主进程在join之后用get把结果捞出来。2.2 join之后再get还是边跑边get上面例子是等所有进程结束再一口气把队列清空。但有一个现实问题如果生产速度大于消费速度队列会积压如果主进程在join死等而子进程在队列写满时阻塞就会死锁。所以更稳的姿势是边跑边取import multiprocessing import time def worker(name, duration, queue): time.sleep(duration) queue.put((name, duration)) if __name__ __main__: q multiprocessing.Queue() procs [] for i in range(5): p multiprocessing.Process(targetworker, args(f任务{i}, i 1, q)) procs.append(p) p.start() # 主进程在这里边收边等 collected [] for _ in range(5): item q.get() # 会阻塞直到有数据 collected.append(item) print(收到:, item) for p in procs: p.join()有几个细节要说破q.get()是阻塞的如果某些进程提前崩了没往队列里写主进程就会卡死在get()。所以实际生产代码里要用q.get(timeout...)或配合task_done来兜底。队列里取出的对象顺序不一定是任务的提交顺序因为多个进程在抢着写队列。如果业务上要求结果按提交顺序对齐就要在任务里带上序号回到主进程再排序。q.empty()这个判断在消费端不太可靠因为进程间数据同步有时间差这边刚接触empty()时返回True下一刻子进程的数据就到了。2.3 我踩过的坑Queue在spawn模式下容易卡一个我印象深刻的坑在Linux下multiprocessing默认用fork创建子进程Queue用起来很顺但把代码挪到Windows或者macOS没有fork时进程改用spawn方式启动此时Queue必须作为参数传给子进程而不是在子进程函数里multiprocessing.Queue()重新创建否则子进程拿到的Queue跟主进程的Queue不是同一个底层队列。还有一次我在一个类方法里创建了Queue传给孩子结果孩子内部又复制了一份数据发出去主进程收不到。排查好久才发现Queue对象不能直接作为普通参数在多进程里传来传去后还指望它在不同进程里指向同一个管道。正确的做法是在顶层创建Queue然后把Queue对象作为args传给子进程函数不要经过中间层包装。3. 用Manager共享对象把结果放进共享dict3.1 Manager究竟做了什么Manager类会在主进程之外另起一个ServerProcess统一管理一组代理对象。子进程通过这些代理对象去修改远端的共享容器而修改会被同步回主进程。你可以理解成主进程开了一个“储物柜”每个进程手里拿的是一张“取件凭证”谁往里存东西主进程都能看见。这个方案的优点是简单、直观特别适合结果是一个字典、列表、嵌套结构比如{任务1: {状态: 成功, 数据: [...]}}这种。3.2 用共享dict收集结果的完整示例import multiprocessing import time def worker(task_id, manager_dict): time.sleep(1) manager_dict[task_id] task_id * task_id if __name__ __main__: with multiprocessing.Manager() as manager: result_dict manager.dict() procs [ multiprocessing.Process(targetworker, args(i, result_dict)) for i in range(1, 6) ] for p in procs: p.start() for p in procs: p.join() print(dict(result_dict))运行结果类似这样{1: 1, 2: 4, 3: 9, 4: 16, 5: 25}注意几点manager.dict()返回的是一个DictProxy代理对象不是普通dict但也支持[]访问和常见字典方法。只有需要跨进程共享的容器才通过Manager创建普通参数照旧用args传。如果结果量极大建议换回Queue或Pool因为Manager有一个后台服务进程做中转性能不如直接的管道通道。3.3 不要高频写Manager对象性能会很难看我做过一个粗糙测试1万个任务同时写一个Manager的dict耗时比Queue方案明显高出一截。原因不难理解每次写操作都要走代理、序列化、网络或管道传输落到ServerProcess那边再更新。这个链路比写入管道要长。所以我的建议是结果数量在几千以内结构复杂用Manager很舒服。结果数量大、结构简单用Queue或Pool任务队列。如果任务是CPU密集且结果需要频繁共享优先考虑multiprocessing.shared_memory或数组类共享内存但那属于另一个深水区这里不展开。4. 用Pool拿到“像返回值一样的结果”4.1 为什么推荐进程池Queue和Manager的方案让你想什么时候创建进程就什么时候创建进程但进程的创建和销毁开销很大。如果一次要跑100个任务手动开100个进程把系统资源直接打满如果分批跑代码又显得啰嗦。Pool就是为此而生——它固定开N个worker进程任务丢进去由空闲worker执行执行完的结果被Pool收集起来。4.2 map和starmap批量任务的返回容器Pool.map的用法跟内置map几乎一样传一个函数和一个可迭代对象返回一个列表列表顺序跟传入顺序一致import multiprocessing import time def calc_square(x): time.sleep(1) return x * x if __name__ __main__: with multiprocessing.Pool(processes4) as pool: result pool.map(calc_square, range(1, 11)) print(result) # [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]这段代码在机器上跑一遍就能感受到多进程的好处10个任务各睡1秒若串行要10秒这里4个worker同时干活大约3秒内完成并且结果是一个列表直接能用。starmap是应对多参数版本的mapfrom multiprocessing import Pool def calc(x, y): return x y if __name__ __main__: with Pool(4) as pool: result pool.starmap(calc, [(1, 2), (3, 4), (5, 6)]) print(result) # [3, 7, 11]这里有一个容易踩的细节Pool.map会阻塞主进程直到所有任务跑完才返回。如果你的业务是“提交任务后继续干别的结果后面对账”不要用map要改用map_async或者apply_async。4.3 apply_async异步拿到Future再collectapply_async是另一个高频用法。它返回一个AsyncResult对象你可以理解为一张“取件凭据”。任务还在跑的时候你可以去做别的事之后调.get()会阻塞直到结果就绪。import multiprocessing import time def work(x): time.sleep(2) return x * 10 if __name__ __main__: with multiprocessing.Pool(3) as pool: results_async [pool.apply_async(work, args(i,)) for i in range(5)] print(任务已全部提交先干点别的...) time.sleep(1) # 再回头取结果 results [r.get(timeout10) for r in results_async] print(results) # [0, 10, 20, 30, 40]这个模式在真实项目里非常常见。比如你要并行抓取一批网页链接提交任务后可以先去渲染页面骨架等用户操作触发再get()结果。AsyncResult还有很多技巧.get(timeout5)最多等5秒超时抛multiprocessing.context.TimeoutError。.ready()轮询任务是否完成。.successful()判断是否成功如果异常返回False。.get()拿到的是函数真正的返回值如果函数抛了异常会在主进程这边重新抛出所以你可以在主进程统一做异常处理。4.4 回调函数任务完成瞬间就处理结果如果不希望主进程反复轮询apply_async还支持callback参数。子进程完成某个任务后Pool的一个worker会在主进程中调用这个回调函数注意回调是在主进程里执行的参数就是任务返回值from multiprocessing import Pool import time def worker(x): time.sleep(2) return x ** 2 def on_done(result): print(f拿到结果: {result}) if __name__ __main__: with Pool(4) as pool: for i in range(8): pool.apply_async(worker, args(i,), callbackon_done) # 不能直接退出with块否则会等待所有任务完成 pool.close() pool.join()这里要特别提醒回调函数本身运行在主进程所以它里面不要再尝试开多进程或者做耗时操作否则整个主进程会卡住。回调适合做轻量级收集比如把结果append到列表中或者发一个进度通知。5. 用concurrent.futures.ProcessPoolExecutor拿返回值5.1 Executor的思路Future模式和Pool异曲同工Python标准库concurrent.futures里的ProcessPoolExecutor在写法上比multiprocessing.Pool更贴近现代并发风格。它把“能不能拿到返回值”这件事抽象成一个Future对象。你提交一个函数立刻得到一个Future之后调用future.result()阻塞取结果或者用as_completed等待一批Future完成。5.2 代码示例submit和mapfrom concurrent.futures import ProcessPoolExecutor, as_completed def heavy_task(n): return n * n if __name__ __main__: tasks range(1, 11) with ProcessPoolExecutor(max_workers4) as executor: futures {executor.submit(heavy_task, n): n for n in tasks} # 按完成顺序输出 for future in as_completed(futures): result future.result() print(f任务参数: {futures[future]}, 结果: {result})executor.map则与Pool.map类似返回一个迭代器按传入顺序产出结果from concurrent.futures import ProcessPoolExecutor def heavy_task(n): return n ** 3 if __name__ __main__: with ProcessPoolExecutor(max_workers4) as executor: for result in executor.map(heavy_task, range(1, 6)): print(result) # 1 8 27 64 125Executor的好处之一是与ThreadPoolExecutor用法完全一致想在线程和进程之间切换只需换类名业务代码几乎不用改。这在开发初期很有优势先用线程跑通逻辑再切到进程拿多核性能。5.3 超时与异常处理Future的result(timeout...)比AsyncResult.get()更早引入API也更规整from concurrent.futures import ProcessPoolExecutor, TimeoutError def slow_task(x): import time time.sleep(5) return x if __name__ __main__: with ProcessPoolExecutor(max_workers2) as executor: future executor.submit(slow_task, 1) try: result future.result(timeout3) except TimeoutError: print(任务超时) future.cancel() # 注意运行中的任务不一定能取消成功注意一个细节主进程任务里抛出异常future.result()会重新抛出那个异常所以你可以直接在主进程这里分类处理。而如果不用Future直接裸写多进程子进程的异常默认只打印到stderr主进程很难感知这也是很多人愿意用Executor的原因。6. 多进程返回值实战中的几条硬经验6.1 所有返回值必须能被pickle不管用Queue、Manager还是Pool只要数据跨进程传输就必须序列化。默认的序列化协议是pickle。你自定义的类的实例、lambda函数、局部内部类常常无法被pickle导致子进程在put结果那一刻直接报PicklingError。这个坑的特点是任务计算得很好结果却在“出门”时被拦下。排查方法很简单在主进程里先手动pickle.dumps(结果)试试。如果主进程都序列化不了子进程更不可能。模块顶层的普通函数和类是可以pickle的__main__模块里定义的函数在Linux fork下还好说在Windows spawn下经常出问题。稳妥套路是把任务函数放在单独模块文件里不要随手写在if __name__ __main__下面。6.2 fork和spawn为什么不同平台行为不一样Linux上multiprocessing默认用fork子进程直接继承父进程的内存快照创建快全局变量直接可见。但它有个著名的坑如果在父进程里已经开了多个线程再用fork子进程可能继承到不健康的锁状态直接死锁。另外fork下子进程里改全局变量不影响父进程容易让人误以为“共享了”。Windows和macOSPython 3.8默认用spawn子进程会重新导入模块、重新创建解释器所以if __name__ __main__这一行保护就非常重要。不写的话Windows上会无限递归地创建子进程。在需要跨平台的业务代码里我的习惯是所有入口逻辑包在main()里并放在if __name__ __main__:下。任务函数放在独立模块中。不使用裸的multiprocessing.Queue()在全局级别创建而是放到main里创建后传入。6.3 大结果集的搬运成本别把所有结果一次性堆内存如果每个进程返回的结果占几百MB把它们全部收集到主进程内存里可能导致OOM。多进程只是把计算分摊了结果收集回来仍然是集中存储。遇到这种场景我倾向于不要让子进程把大结果全量返回而是子进程直接计算结果并写入磁盘或数据库只向主进程返回一个文件路径或记录ID。或者用共享内存数组这种方案需要固定长度、提前分配内存适合数值密集型任务。还有个经常被忽略的问题Queue里堆积太多对象时会因为内存压力反过来拖慢所有进程。所以真正跑离线批量任务时我会在主进程里主动消费队列而不是等任务全部结束再一次性读。6.4 关于进程数选择的苦口婆心开多少进程不是越多越好。CPU密集型任务开os.cpu_count()或者减一两个留点余量一般就够了如果任务里还涉及大量IO或者等待外部接口可以把进程数适当放大但不要超过cpu_count * 2。开太多进程后操作系统的进程切换开销反而超过了并行收益。测过一组对比同样的8核心机器跑10万个CPU密集任务4进程耗时约5.2秒8进程耗时约3.1秒16进程反而回到了3.8秒。原因就是进程切换和各自上下文开销。所以不是无脑堆进程数。7. 说在最后场景与方案的匹配多进程返回值获取这件事没有银弹。我个人的选择习惯是任务量少、只要简单结果Queue或Manager都行哪个顺手用哪个。大量独立任务需要批量收集结果首选Pool.map或ProcessPoolExecutor.map。任务提交后主进程要干别的、结果后面对账选apply_async或Executor.submit配合as_completed。结果结构复杂但量不大用Manager共享dict最省心。结果数据量巨大让子进程直接落盘或写库主进程只拿处理完任务的通知。回到最初的问题“多进程返回值怎么拿”答案是没法直接拿但你可以通过队列、共享对象、进程池Future机制去拿。理解了这一层之后剩下的就是根据数据量、任务量、平台的限制去选一条最顺手的路。下次再遇到进程结果丢失先别怀疑写错了函数先想想结果是不是被你放在了只有子进程自己知道的地方。