企业官网建设流程全解析

如果一个项目里有大量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(target=my_func)创建出的子进程，是一份完整的内存快照拷贝（fork时是写时复制）或者全新解释器（spawn时），子进程里所有变量跟主进程没有半毛钱关系。你在子进程里写了return 42，这个42只存在于子进程自己的调用栈里，子进程退出，栈被销毁，主进程什么都没得到。所以“子进程函数里写return，主进程想直接接住”这件事，从设计上就不成立。

用生活类比就是：线程是一家人住一间房，你喊一声全家都听见；进程是几个人各住各的房间，你喊破嗓子隔壁也听不见，必须用电话线（IPC机制）才能传话。

1.2 return在进程里到底去哪了

很多人试过这样写：

from multiprocessing import Process def compute(x): return x * 2 p = Process(target=compute, 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(target=worker, args=("任务A", 2, q)), multiprocessing.Process(target=worker, 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(target=worker, 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共享对象：把结果放进共享dict

3.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(target=worker, 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(processes=4) 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是应对多参数版本的map：

from 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，再collect

apply_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(timeout=10) for r in results_async] print(results) # [0, 10, 20, 30, 40]

这个模式在真实项目里非常常见。比如你要并行抓取一批网页链接，提交任务后可以先去渲染页面骨架，等用户操作触发再get()结果。

AsyncResult还有很多技巧：

• .get(timeout=5)：最多等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,), callback=on_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和map

from concurrent.futures import ProcessPoolExecutor, as_completed def heavy_task(n): return n * n if __name__ == "__main__": tasks = range(1, 11) with ProcessPoolExecutor(max_workers=4) 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_workers=4) as executor: for result in executor.map(heavy_task, range(1, 6)): print(result) # 1 8 27 64 125

Executor的好处之一是与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_workers=2) as executor: future = executor.submit(slow_task, 1) try: result = future.result(timeout=3) 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和macOS（Python 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机制去拿。理解了这一层之后，剩下的就是根据数据量、任务量、平台的限制去选一条最顺手的路。下次再遇到进程结果丢失，先别怀疑写错了函数，先想想结果是不是被你放在了只有子进程自己知道的地方。