Python多处理模块multiprocessing实战指南：进程创建与数据通信

周予安

发布时间：2026-10-12 | 2人浏览过

原创

本文详解 Python 多处理模块中 Process 类的标准用法。针对 CPU 密集型任务受全局解释器锁限制的问题，提供从零创建子进程、批量并发执行、跨进程数据传递及异常隔离处理的完整实操指南，帮助开发者快速构建高效的多进程并行计算流程。

核心机制与适用场景

multiprocessing 模块提供了与 threading 相似的 API 来生成独立进程。其核心优势在于完全绕开 Python 的全局解释器锁（GIL），适用于需要真正并行执行的 CPU 密集型场景。与共享内存的线程不同，每个多处理对象都在独立的地址空间中运行，资源隔离性强但创建开销较大。

基础操作：创建与管理单个子进程

在 Python 中运行任何脚本都会自动成为父进程。要调度额外任务，需使用 Process 类包装目标函数，并通过控制方法管理生命周期。

第1步：定义子任务函数

子任务函数应为顶层可导入对象，避免在函数内部隐式依赖全局状态。以下示例演示如何识别当前进程 ID：

import os

def worker_task():
    print(f"Hello! 我是子进程 {os.getpid()}")

if __name__ == "__main__":
    print(f"Hello! 我是主进程 {os.getpid()}")

第2步：实例化并启动进程

将目标函数赋值给 target 参数即可创建 Process 实例。调用 start() 后，操作系统会派生新进程并执行 run() 逻辑：

from multiprocessing import Process

def worker_task():
    print(f"Hello! 我是子进程 {os.getpid()}")

if __name__ == "__main__":
    print(f"Hello! 我是主进程 {os.getpid()}")
    
    sub_process = Process(target=worker_task)
    sub_process.start()

第3步：同步等待进程结束

直接使用 start() 属于异步操作，主程序会继续向下执行。若需确保主进程在子任务完成后再接收结果或清理资源，必须调用 join()。该方法会使主进程阻塞直至子进程终止：

sub_process.join()
print("子任务已执行完毕")

若省略 join()，主程序可能提前退出导致僵尸进程，或后续代码在子进程尚未就绪时读取空数据，引发竞态条件。

进阶应用：批量创建与参数传递

当需要并行处理多项独立任务时，可通过循环批量实例化，并利用 args 元组向子函数注入参数。注意列表需预先存储实例引用，以免作用域回收导致进程意外终止：

from multiprocessing import Process

def task_runner(task_id):
    print(f"进程 {os.getpid()} 正在处理任务 #{task_id}")

if __name__ == "__main__":
    process_pool = []
    for i in range(5):
        p = Process(target=task_runner, args=(i,))
        process_pool.append(p)
    
    for p in process_pool:
        p.start()
    
    for p in process_pool:
        p.join()

跨进程数据通信机制

由于进程间内存严格隔离，子函数无法通过 return 直接将值交还给父进程。正确的做法是使用 Queue 作为 IPC（进程间通信）通道。队列实现了先进先出协议，并内置了底层的管道与锁机制：

from multiprocessing import Process, Queue

def compute_data(q, val1, val2):
    result = val1 + val2
    q.put(result)  # 写入队列

if __name__ == "__main__":
    data_queue = Queue()
    proc = Process(target=compute_data, args=(data_queue, 12, 8))
    
    proc.start()
    proc.join()
    
    fetched_val = data_queue.get()  # 读取队列
    print(f"获取到计算结果: {fetched_val}")

子进程异常处理策略

Process 类在设计上具有异常隔离特性：子进程中抛出的异常默认仅打印至标准错误流，不会向上冒泡触发父级的 try...except。若父进程需要感知失败状态，必须在子任务内部完成捕获：

def safe_worker():
    try:
        # 模拟高风险操作
        raise ValueError("数据解析失败")
    except Exception as e:
        print(f"子进程已捕获异常: {e}")

if __name__ == "__main__":
    p = Process(target=safe_worker)
    p.start()
    p.join()
    print("主进程继续安全运行")