知识花园

multiprocessing 多进程

5分钟阅读

multiprocessing 是 Python 标准库的多进程模块，用进程替代线程绕过 GIL 限制实现真正并行；跨平台与嵌套场景首选 spawn 启动方式。

官方文档：https://docs.python.org/3/library/multiprocessing.html

核心概念

  • 进程：独立运行环境，拥有独立内存空间
  • GIL：全局解释器锁，同一时刻只允许一个线程执行字节码
  • multiprocessing：使用进程替代线程，实现真正并行

基本用法

创建进程

import multiprocessing
 
def worker(name):
    print(f"Worker {name} running")
 
if __name__ == "__main__":
    p1 = multiprocessing.Process(target=worker, args=("A",))
    p2 = multiprocessing.Process(target=worker, args=("B",))
 
    p1.start()
    p2.start()
    p1.join()
    p2.join()

进程池

from multiprocessing import Pool
 
def square(x):
    return x * x
 
if __name__ == "__main__":
    with Pool(4) as p:
        results = p.map(square, [1, 2, 3, 4, 5])
    print(results)  # [1, 4, 9, 16, 25]

进程间通信

# 队列
queue = multiprocessing.Queue()
queue.put(item)
item = queue.get()
 
# 共享内存
counter = multiprocessing.Value('i', 0)  # 整数
arr = multiprocessing.Array('i', [1, 2, 3])  # 数组

共享内存示例

def increment(counter):
    for _ in range(1000):
        counter.value += 1
 
if __name__ == "__main__":
    counter = multiprocessing.Value('i', 0)
    processes = [multiprocessing.Process(target=increment, args=(counter,)) for _ in range(4)]
 
    for p in processes:
        p.start()
    for p in processes:
        p.join()
 
    print(f"Counter: {counter.value}")

启动方式：fork vs spawn

spawn 是跨平台的「干净启动」，fork 是 Unix 的「快速复制」。

维度fork（Unix 专属）spawn（跨平台）
创建机制复制父进程地址空间（COW）启动新解释器，重新导入脚本
启动速度极快较慢
资源继承继承所有资源仅继承必要资源
线程安全父进程有活跃线程时易死锁天然线程安全
嵌套进程易出问题完美支持
跨平台差（Windows 不支持）好（全平台通用）

原理：

  • fork：调用 fork() 复制父进程整个地址空间，子进程从 fork() 位置开始执行；写时复制（COW），修改数据时才真正拷贝内存
  • spawn：启动全新 Python 解释器，子进程重新导入当前脚本，因此必须用 if __name__ == '__main__': 保护主逻辑

实践验证：全局变量的差异

fork 模式子进程直接继承父进程的全局变量；spawn 模式子进程重新执行模块顶层代码，全局变量回到初始状态：

import multiprocessing
 
GLOBAL_VAR = "父进程初始化"
 
def worker():
    print(f"子进程: {GLOBAL_VAR}")
 
if __name__ == '__main__':
    # fork：继承父进程的全局变量
    multiprocessing.set_start_method('fork')
    p = multiprocessing.Process(target=worker)
    p.start(); p.join()
 
    # spawn：重新初始化全局变量
    multiprocessing.set_start_method('spawn', force=True)
    p = multiprocessing.Process(target=worker)
    p.start(); p.join()

嵌套进程

子进程里可以再启动子进程。关键三点：

  1. 主进程设置 multiprocessing.set_start_method()，重复调用会报错，放脚本开头或用 force=True
  2. 用 if __name__ == '__main__': 保护入口
  3. Windows 必须用 spawn；fork 模式嵌套大概率死锁，spawn 稳定运行
import multiprocessing
import time
 
def nested_worker(num):
    print(f"嵌套进程 {num} PID: {multiprocessing.current_process().pid}")
    time.sleep(1)
 
def parent_worker(pid):
    print(f"父进程 {pid} PID: {multiprocessing.current_process().pid}")
    processes = []
    for i in range(2):
        p = multiprocessing.Process(target=nested_worker, args=(f"{pid}-{i}",))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
 
if __name__ == '__main__':
    multiprocessing.set_start_method('spawn', force=True)
    main_processes = []
    for i in range(2):
        p = multiprocessing.Process(target=parent_worker, args=(i,))
        main_processes.append(p)
        p.start()
    for p in main_processes:
        p.join()

嵌套进程用 join() 确保生命周期可控，层级限制在 2-3 层，并用 Pool 限制并发数。

避坑指南

问题原因解决
Windows 下 RuntimeError未加 if __name__ == '__main__': 保护所有进程启动逻辑放保护块内
fork 模式死锁父进程有活跃线程或嵌套进程改用 spawn
spawn 启动慢每次新建解释器用进程池复用
set_start_method 报错重复调用放脚本开头，或用 force=True
资源耗尽嵌套层级过多限制 2-3 层，使用 Pool
进程间通信错误直接共享变量使用 Queue/Pipe/Value/Array

此外：

  • 进程创建开销比线程大，适合 CPU 密集型任务；I/O 密集型优先考虑 asyncio
  • 进程间共享数据必须走 Queue、Value、Array 等原语

选择建议

场景推荐
纯 Linux/macOS、无嵌套、追求性能fork
跨平台、嵌套进程、有活跃线程spawn
数据处理、进程池spawn
简单脚本spawn（减少适配成本）

关系图谱

反向链接