在 Python 中，使用 `multiprocessing` 模块实现多进程时，**进程间传递参数**本质上是通过**进程启动时的参数传递机制**完成的（如 `Process(target=func, args=..., kwargs=...)`），但需特别注意：**进程间内存不共享**，因此传递的参数会被序列化（pickle）后拷贝到子进程中 —— 这意味着： - ✅ 可以安全传递 **不可变对象**（`int`, `str`, `tuple`, `frozenset`）、**可序列化对象**（`list`, `dict`, `numpy.ndarray`（若支持 pickle）、自定义类实例（满足 pickle 条件）等）； - ⚠️ 不能直接传递 **不可 pickle 的对象**（如 `lambda` 函数、嵌套函数、带绑定方法的未绑定实例、`threading.Lock`、文件对象、生成器、某些 C 扩展对象等）； - ❌ 不能通过“全局变量”或“闭包”隐式共享数据（子进程有独立内存空间，修改不会影响父进程）。 --- ### ✅ 正确传递参数的常用方式（4 种典型场景） #### 1. 基础方式：`Process` + `args`/`kwargs` ```python import multiprocessing as mp import time def worker(name: str, count: int, flag: bool = False): print(f"[{mp.current_process().name}] Hello {name}, count={count}, flag={flag}") time.sleep(1) if __name__ == "__main__": # 启动多个进程，显式传参 p1 = mp.Process(target=worker, args=("Alice", 5), kwargs={"flag": True}) p2 = mp.Process(target=worker, args=("Bob", 3), kwargs={"flag": False}) p1.start() p2.start() p1.join() p2.join() ``` > ✅ 安全、清晰、推荐用于简单参数传递。 > ⚠️ 注意：`args` 是 tuple，单个元素记得加逗号：`args=("Alice",)`，否则 `"Alice"` 被当字符串迭代 → `(‘A’,‘l’,‘i’,‘c’,‘e’)`！ --- #### 2. 使用 `multiprocessing.Pool` 传递参数（适用于批量同构任务） ```python from multiprocessing import Pool import os def square_with_id(x): return os.getpid(), x * x if __name__ == "__main__": # 传入可迭代对象，每个元素作为单独参数调用函数 with Pool(3) as pool: # 方式1：map —— 单参数函数 results = pool.map(square_with_id, [1, 2, 3, 4, 5]) print(results) # 输出示例: [(12345, 1), (12346, 4), (12345, 9), ...] # 方式2：starmap —— 多参数解包（推荐！） def power(base, exp): return base ** exp tasks = [(2, 3), (3, 2), (5, 2)] # 每个元组对应 base, exp with Pool(2) as pool: results = pool.starmap(power, tasks) print(results) # [8, 9, 25] ``` > ✅ `starmap` 是处理多参数最自然的方式（避免写 wrapper）。 > ⚠️ `Pool` 中函数必须是**模块级顶层函数**（不能是局部函数/lambda），否则 pickle 失败。 --- #### 3. 传递复杂对象（自定义类、NumPy 数组等） ```python import multiprocessing as mp import numpy as np class Config: def __init__(self, model_name: str, lr: float): self.model_name = model_name self.lr = lr def train_worker(config: Config, data: np.ndarray, epoch: int): print(f"Training {config.model_name} (lr={config.lr}) on shape {data.shape}, epoch {epoch}") # 模拟训练 return data.sum() * epoch if __name__ == "__main__": config = Config("ResNet", 0.001) X = np.random.randn(1000, 784) # 模拟数据 # ✅ NumPy array 和自定义类（无不可 pickle 属性）均可安全传递 p = mp.Process( target=train_worker, args=(config, X, 10) ) p.start() p.join() ``` > ✅ `np.ndarray` 默认可 pickle（底层数据被序列化）；若数组极大，建议用 `shared_memory` 或 `Array`/`RawArray` 避免拷贝（见下文进阶）。 > ⚠️ 若 `Config` 包含 `threading.Lock`、`open()` 文件句柄等，则会报 `TypeError: can't pickle _thread.lock objects`。 --- #### 4. 进阶：共享内存（避免大数组拷贝）→ 使用 `multiprocessing.Array` / `multiprocessing.shared_memory` ```python import multiprocessing as mp import numpy as np from multiprocessing import shared_memory import time def worker_shm(shm_name: str, shape: tuple, dtype: str, idx: int): # 重新连接共享内存 existing_shm = shared_memory.SharedMemory(name=shm_name) # 重建 numpy 数组（指向共享内存） arr = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf) result = arr.sum() * (idx + 1) print(f"Worker {idx}: sum={arr.sum():.2f} → result={result}") existing_shm.close() if __name__ == "__main__": # 创建共享内存（大数组） arr = np.random.randn(10_000, 1000) # ~80MB shm = shared_memory.SharedMemory(create=True, size=arr.nbytes) shared_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf) shared_arr[:] = arr[:] # 复制数据进去 # 启动多个进程，只传 shm.name、shape、dtype 等轻量参数 processes = [] for i in range(3): p = mp.Process(target=worker_shm, args=(shm.name, arr.shape, arr.dtype, i)) processes.append(p) p.start() for p in processes: p.join() # 清理 shm.close() shm.unlink() # 删除共享内存 ``` > ✅ 极大提升大数据场景性能（零拷贝）； > ⚠️ 需手动管理生命周期（`close()` + `unlink()`），且需同步（如加 `Lock`）防止竞态。 --- ### ❌ 常见错误及原因 | 错误现象 | 原因 | 解决方案 | |----------|------|-----------| | `TypeError: can't pickle _thread.lock objects` | 尝试传递 `threading.Lock`、`logging.Logger` 等不可序列化对象 | 改用 `multiprocessing.Lock`，或仅传基础参数，在子进程中新建对象 | | `AttributeError: Can't get attribute 'xxx' on <module '__main__'>` | Windows/macOS 上 `spawn` 启动方式要求主模块可导入；函数不在顶层或 `if __name__ == "__main__":` 缺失 | 确保入口保护，函数定义在模块顶层，Windows 必须加 `if __name__ == "__main__":` | | 子进程读取不到参数修改 | 误以为修改 `args` 中的 list/dict 会影响父进程 | 进程隔离 → 改用 `Manager().list()` 或 `Queue`/`Pipe` 显式通信 | | `OSError: [Errno 24] Too many open files` | 大量进程+大量文件描述符（如每个进程打开日志文件） | 限制进程数、复用资源、用 `concurrent.futures.ProcessPoolExecutor(max_workers=...`) 控制 | --- ### 🔑 关键总结 | 场景 | 推荐方式 | 是否拷贝数据 | 备注 | |------|-----------|----------------|------| | 简单参数（int/str/list/dict） | `Process(args=..., kwargs=...)` 或 `Pool.starmap()` | ✅ 是（pickle 拷贝） | 最常用、最安全 | | 大 NumPy 数组 | `shared_memory` + `ndarray` | ❌ 否（共享内存） | 需手动管理，适合 >10MB 数据 | | 需要子进程返回结果 | `Queue`, `Pipe`, `Manager().dict()` | ✅/❌ 按需 | `Queue` 最易用；`Manager` 开销大但接口友好 | | 需要同步（如计数器） | `multiprocessing.Value`, `multiprocessing.Lock` | ❌ 共享 | 非 pickle，专为跨进程设计 | ---