Python并发04：多进程加多协程混合模式的分层并发设计

多进程 + 多协程：混合模式的分层并发设计

很多真实任务既有 I/O 等待，又有 CPU 计算。只用多进程时，进程会在等待 I/O 时闲置；只用多协程时，计算部分仍受 GIL 限制。混合模式把两者组合起来：外层用进程池获得多核并行能力，内层用协程池重叠 I/O 等待。

一、为什么需要混合模式

数据管道类任务通常由三步组成：

1. 从数据库或网络异步读取数据，I/O 密集。
2. 对读取到的数据做复杂统计计算，CPU 密集。
3. 把结果异步写回数据库，I/O 密集。

如果只用多进程，每个进程都必须等待 I/O，CPU 会空闲。如果只用协程，计算阶段仍然无法利用多核。因此可以用“外层多进程、内层多协程”的方式各取所长。

二、双层架构与职责划分

可以把混合模式想象成一家现代化工厂：

• 外层进程池相当于多个车间，每个车间独立运行，数量通常接近 CPU 核心数。
• 内层协程池相当于车间里的流水线工人，负责处理 I/O 等待和轻量计算。

例如处理 1 亿条用户行为日志：

• 将数据分成 4 份，每份交给一个进程。
• 每个进程启动 3 个协程，分别负责读取、清洗和写入。
• 某个协程等待数据时，其他协程继续计算。

这样既能绕开 GIL，又不会让 I/O 等待浪费 CPU。

三、数据如何流转

进程之间的数据传递要尽量轻量：

四、优势互补

适合混合模式的典型任务包括：

• 数据管道：读、算、存。
• 批量报表：拉取数据、聚合计算、写入结果。
• 高并发后台服务：接请求、查库、计算、返回。

五、混合模式示例

下面的示例把 1 到 4000 分成 4 个数据分片，外层进程池并行处理，内层每个进程用 3 个协程处理异步 I/O 和计算。

importmultiprocessingimportasyncioimporttimeasyncdefasync_worker(task_id,data_chunk,process_name):""" 进程内的协程：处理异步 I/O 与轻量计算 """print(f"[{time.strftime('%X')}]{process_name}- 协程{task_id}：开始处理数据分片（长度：{len(data_chunk)}）")awaitasyncio.sleep(1)result=sum(x*xforxindata_chunk)print(f"[{time.strftime('%X')}]{process_name}- 协程{task_id}：处理完成，结果：{result}")returnresultasyncdefprocess_inner_logic(data_chunk,process_name):""" 单个进程的内部逻辑：创建协程池，处理当前进程的数据分片 """coroutines=[async_worker(i,data_chunk,process_name)foriinrange(3)]coro_results=awaitasyncio.gather(*coroutines)returnsum(coro_results)defprocess_worker(data_chunk):""" 进程工作函数：初始化事件循环，执行进程内协程逻辑 """process_name=multiprocessing.current_process().namereturnasyncio.run(process_inner_logic(data_chunk,process_name))defmain():print("=== Python 混合模式（多进程 + 多协程）===")start_time=time.time()data_chunks=[list(range(1,1001)),list(range(1001,2001)),list(range(2001,3001)),list(range(3001,4001)),]withmultiprocessing.Pool(processes=2)aspool:process_results=pool.map(process_worker,data_chunks)final_result=sum(process_results)end_time=time.time()total_duration=end_time-start_timeprint("\n=== 混合模式任务执行完成 ===")print(f"各进程计算结果：{process_results}")print(f"最终总结果（所有分片平方和之和）：{final_result}")print(f"总耗时：{total_duration:.2f}秒")pure_process_duration=4*1+0.5print(f"若用纯多进程（无协程），理论耗时≈{pure_process_duration:.2f}秒（混合模式效率提升≈{pure_process_duration/total_duration:.1f}倍）")if__name__=="__main__":main()

在 2 核 CPU 环境中，示例输出可能如下：

=== Python 混合模式（多进程 + 多协程）=== [17:00:00] SpawnPoolWorker-1 - 协程 0：开始处理数据分片（长度：1000） [17:00:00] SpawnPoolWorker-1 - 协程 1：开始处理数据分片（长度：1000） [17:00:00] SpawnPoolWorker-1 - 协程 2：开始处理数据分片（长度：1000） [17:00:00] SpawnPoolWorker-2 - 协程 0：开始处理数据分片（长度：1000） [17:00:00] SpawnPoolWorker-2 - 协程 1：开始处理数据分片（长度：1000） [17:00:00] SpawnPoolWorker-2 - 协程 2：开始处理数据分片（长度：1000） [17:00:01] SpawnPoolWorker-1 - 协程 0：处理完成，结果：333833500 [17:00:01] SpawnPoolWorker-1 - 协程 1：处理完成，结果：333833500 [17:00:01] SpawnPoolWorker-1 - 协程 2：处理完成，结果：333833500 [17:00:01] SpawnPoolWorker-2 - 协程 0：处理完成，结果：1667167500 [17:00:01] SpawnPoolWorker-2 - 协程 1：处理完成，结果：1667167500 [17:00:01] SpawnPoolWorker-2 - 协程 2：处理完成，结果：1667167500 [17:00:01] SpawnPoolWorker-1 - 协程 0：开始处理数据分片（长度：1000） [17:00:01] SpawnPoolWorker-1 - 协程 1：开始处理数据分片（长度：1000） [17:00:01] SpawnPoolWorker-1 - 协程 2：开始处理数据分片（长度：1000） [17:00:01] SpawnPoolWorker-2 - 协程 0：开始处理数据分片（长度：1000） [17:00:01] SpawnPoolWorker-2 - 协程 1：开始处理数据分片（长度：1000） [17:00:01] SpawnPoolWorker-2 - 协程 2：开始处理数据分片（长度：1000） [17:00:02] SpawnPoolWorker-1 - 协程 0：处理完成，结果：4501500500 [17:00:02] SpawnPoolWorker-1 - 协程 1：处理完成，结果：4501500500 [17:00:02] SpawnPoolWorker-1 - 协程 2：处理完成，结果：4501500500 [17:00:02] SpawnPoolWorker-2 - 协程 0：处理完成，结果：9335834500 [17:00:02] SpawnPoolWorker-2 - 协程 1：处理完成，结果：9335834500 [17:00:02] SpawnPoolWorker-2 - 协程 2：处理完成，结果：9335834500 === 混合模式任务执行完成 === 各进程计算结果：[1001500500, 5001502500, 13504501500, 28007503500] 最终总结果（所有分片平方和之和）：47515008000 总耗时：2.15 秒 若用纯多进程（无协程），理论耗时≈4.50 秒（混合模式效率提升≈2.1 倍）

从结果可以看出两层并发同时生效：外层进程并行处理分片，内层协程在 1 秒 I/O 等待期间继续调度其他任务。

六、混合模式适用场景

小结

混合模式不是简单叠加两种技术，而是分层解决不同瓶颈：进程层突破 GIL，协程层重叠 I/O。实现时需要控制进程数量、分片均衡和进程间数据传递的体积。