Python多进程编程
发布于
单核算不动，就上多核；多线程卡在 GIL，就上多进程。
这一章解决「算」的问题：把 CPU 密集型任务拆开分给多个进程，
每个进程独占一个核心、绕过 GIL，真正并行。但并行不是免费的午餐——进程独立内存、通信靠序列化、创建开销大。
什么时候该用、怎么用得对、有哪些坑，都在这一章讲清楚。
前置知识
- Python 函数与类的基础用法
核心内容
进程基础概念
进程 vs 程序 vs 线程
- 程序：磁盘上的静态二进制文件（.py/exe），无资源、不运行。
- 进程：程序加载到内存后独立运行实例，操作系统分配独立资源：
- 独立内存空间、CPU 时间片、文件描述符、PID 进程号
- 进程间默认内存隔离，无法直接共享变量
- 线程：进程内轻量级执行单元，共享进程内存，切换开销远小于进程。
为什么需要多进程：GIL 与真并行
GIL（Global Interpreter Lock）是 CPython 的一把全局锁，保证同一时刻只有一个线程执行 Python 字节码。多线程开再多，算力上限也只有一个核。
多进程不同：每个进程有自己独立的解释器和 GIL，进程 A 和进程 B 的计算可以真正发生在不同核心上。
时间线（两个进程各自持有一把 GIL）
──────────────────────────────────────────
进程A: [计算] [计算] [计算] [完成]
进程B: [计算] [计算] [计算] [完成]
──────────────────────────────────────────
两条路径同时推进 → 多进程 = 真并行结论一句话
CPU 密集型任务：多线程受 GIL 限制 ≈ 单核；多进程独占多核 ≈ 成倍加速。
为什么全局变量不共享（内存隔离）
每个子进程启动时，操作系统给它复制一份主进程的内存空间，之后各干各的。子进程里修改全局变量，只改自己那份拷贝，父进程完全看不见。
import multiprocessing
num = 0
def add():
global num
num += 100 # 改的是子进程自己的拷贝
if __name__ == "__main__":
p = multiprocessing.Process(target=add)
p.start()
p.join()
print(num) # 0，父进程的 num 没变多进程共享 ≠ 多线程共享
多线程共享全局变量，要加锁防竞争；多进程的变量根本不共享，
想跨进程共享数据必须走 IPC（Queue / Pipe / 共享内存），见下一章节。
创建进程的两种方式
方式1：函数式（Process + target）
import multiprocessing
import time
def task(name, delay):
"""模拟耗时任务"""
print(f"进程 {name} 启动")
time.sleep(delay)
print(f"进程 {name} 执行完毕")
if __name__ == "__main__":
start = time.perf_counter()
# 创建两个子进程
p1 = multiprocessing.Process(target=task, args=("A", 2))
p2 = multiprocessing.Process(target=task, kwargs={"name": "B", "delay": 1})
# 启动
p1.start()
p2.start()
# 等待两个子进程结束
p1.join()
p2.join()
elapsed = time.perf_counter() - start
print(f"所有子进程执行完成，耗时 {elapsed:.1f}s")
# 串行需要 3s，并行只需 2s
if __name__ == "__main__"必须加
Windows 下创建子进程采用 spawn 方式，会重新导入主模块。
没有这层保护，子进程会再次执行创建进程的代码，无限递归直到崩溃。
原因在「踩坑篇」展开。
方式2：类继承（继承 Process，重写 run）
import multiprocessing
import time
class MyProcess(multiprocessing.Process):
def __init__(self, name, delay):
super().__init__(name=name)
self.delay = delay
def run(self):
print(f"进程 {self.name} 启动")
time.sleep(self.delay)
print(f"进程 {self.name} 结束")
if __name__ == "__main__":
p1 = MyProcess("进程1", 1)
p2 = MyProcess("进程2", 2)
p1.start()
p2.start()
p1.join()
p2.join()
print("程序结束") # 逐个等待，全部跑完再继续
join()不能省
主进程代码执行完毕后，解释器会检查是否还有非守护的子进程在运行：
有 → 继续等待；没有 → 退出，守护子进程被强制销毁。
想拿到子进程的成果再往下走，就必须join()等它结束。
补充：守护进程
import multiprocessing
import time
def daemon_task():
while True:
print('守护进程正在运行中...')
time.sleep(1)
def normal_task():
time.sleep(3)
print('普通子进程结束')
if __name__ == '__main__':
# daemon=True：主进程结束，守护进程直接被回收
p1 = multiprocessing.Process(target=daemon_task, daemon=True)
# p2 不是守护进程，主进程要等它结束
p2 = multiprocessing.Process(target=normal_task)
p1.start()
p2.start()
p2.join()
print('主进程执行完毕，守护进程直接销毁...')运行逻辑：3 秒后普通进程结束，主进程结束，无限循环的守护进程直接终止。
daemon必须在start()前设置，启动后修改无效
守护进程不能执行文件、数据库落盘操作，主进程退出会强制杀死，数据丢失。
补充：进程常用属性与方法
| 方法 / 属性 | 作用 |
|---|---|
start() | 启动进程 |
run() | 进程执行逻辑入口 |
join(timeout=None) | 阻塞主进程等待子进程，timeout 设最大等待秒数 |
is_alive() | 判断进程是否正在运行 |
name | 进程名称 |
pid | 进程号（线程没有） |
terminate() | 强制终止进程（发 SIGTERM） |
kill() | 直接杀死进程（发 SIGKILL，无法捕获） |
multiprocessing.current_process() | 获取当前进程对象 |
os.getpid() / os.getppid() | 获取当前 / 父进程 PID |
terminate / kill 的区别
terminate()发 SIGTERM，进程可捕获做清理；kill()发 SIGKILL，
操作系统直接干掉，无法被捕获，finally里的清理代码都可能不执行。
进程间通信（IPC）
通信方式速览（表格：Queue / Pipe / Value / Manager）
| 方式 | 数据 | 性能 | 适合场景 | 一句话 |
|---|---|---|---|---|
Queue | 任意可 pickle | 中 | 生产消费、任务分发 | 默认首选 |
Pipe | 任意可 pickle | 快 | 两个进程直接通信 | 一对一 |
Value / Array | 基本类型 / 数组 | 最快 | 高频更新简单数据 | 必须配 Lock |
Manager | 复杂对象 | 慢 | 共享 dict/list 等 | 慎用 |
一句话推荐
不知道用哪个，就用Queue。
队列 Queue（推荐默认方案）
multiprocessing.Queue 是进程安全的队列：内部用管道 + 锁实现，支持多生产者、多消费者，不用担心竞争问题。
put(item)：放入数据，队列满时阻塞get()：取出数据，队列空时阻塞task_done()：标记一个任务完成（配合join()使用）join()：阻塞直到所有任务被标记完成
生产消费模型：
import multiprocessing
import time
def producer(q):
for i in range(5):
q.put(f"任务{i}")
print(f"生产：任务{i}")
q.put(None) # 哨兵值：通知消费者退出
def consumer(q):
while True:
item = q.get()
if item is None: # 收到哨兵，退出
q.put(None) # 传给下一个消费者
break
print(f"消费：{item}")
time.sleep(0.2)
if __name__ == "__main__":
q = multiprocessing.Queue()
ps = [multiprocessing.Process(target=producer, args=(q,)) for _ in range(2)]
cs = [multiprocessing.Process(target=consumer, args=(q,)) for _ in range(2)]
for p in ps + cs:
p.start()
for p in ps + cs:
p.join()
print("全部完成")队列里的数据必须能 pickle
放进队列的对象会被序列化后在进程间传输，不可序列化的对象（如锁、
套接字）会报错。详见「序列化限制：pickle」。
管道 Pipe
Pipe() 返回一对连接对象，默认是双向（duplex=True）：两端都可以 send 和 recv。通过 duplex=False 可以创建单向管道。
import multiprocessing
def send(conn):
conn.send("你好")
conn.close()
if __name__ == "__main__":
parent_conn, child_conn = multiprocessing.Pipe() # 默认双向
p = multiprocessing.Process(target=send, args=(child_conn,))
p.start()
print(parent_conn.recv()) # 你好
p.join()和 Queue 的区别：
| Pipe | Queue | |
|---|---|---|
| 通信方 | 一对一（两端） | 多生产者 / 多消费者 |
| 安全 | 原始，无内置锁 | 内置锁 + 缓冲区，安全 |
| 性能 | 更快 | 稍慢 |
| 使用 | 需自己管理协议 | 开箱即用 |
不要在一端同时 send / recv
管道是阻塞式的，同一端同时收发可能互相等待，导致死锁。
设计好「谁发谁收」，避免双向同时操作。
共享内存：Value / Array / SharedMemory
Value / Array 把数据放进操作系统共享内存，所有子进程直接读写，不需要序列化，速度最快。
Value(type, init)：单个值，type 是 ctypes 类型编码，如'i'(int)、'd'(float)Array(type, size_or_list)：定长数组
import multiprocessing
def increment(num, lock):
for _ in range(10_000):
with lock: # 读改写必须加锁，否则结果偏小
num.value += 1
if __name__ == "__main__":
lock = multiprocessing.Lock()
num = multiprocessing.Value("i", 0)
ps = [multiprocessing.Process(target=increment, args=(num, lock)) for _ in range(4)]
for p in ps:
p.start()
for p in ps:
p.join()
print(num.value) # 40000，不加锁会小于这个值为什么必须配合 Lock
num.value += 1是「读 → 加 → 写回」三步，不是原子操作。
多个进程同时执行会互相覆盖，结果偏小。共享内存必须用Lock保护。
multiprocessing.Value可传lock=False关闭内置锁，手动管理，性能更高但更危险。
和 Queue 的性能对比：
| 共享内存 | Queue / Pipe | |
|---|---|---|
| 传输 | 直接读写内存 | 每次都要 pickle 序列化 |
| 速度 | 最快 | 慢一个数量级 |
| 适合 | 高频更新的简单数据 | 低频、复杂、对象数据 |
Python 3.8+ 的
shared_memory.SharedMemory
可以共享任意大小的字节块，适合大数组（配合 numpy）。但它没有内置锁
和同步，用起来更底层，适合追求极致性能的场景。
Manager：共享对象服务器
Manager 启动一个独立的服务器进程，所有子进程通过代理对象访问服务器上维护的数据结构（dict / list / Namespace 等）。
import multiprocessing
def worker(d, key, value):
d[key] = value # 通过代理写服务器上的字典
if __name__ == "__main__":
with multiprocessing.Manager() as manager:
d = manager.dict()
ps = [multiprocessing.Process(target=worker, args=(d, i, i * 10)) for i in range(4)]
for p in ps:
p.start()
for p in ps:
p.join()
print(d) # {0: 0, 1: 10, 2: 20, 3: 30}性能开销大，慎用
每次访问代理对象都要经过一次 IPC，性能远低于 Queue 和共享内存。
适合「低频读写复杂结构」，不适合高频热路径。
序列化限制：pickle
进程间传数据，本质是把对象序列化（pickle）成字节流，到另一端再还原。所以能传什么、不能传什么，取决于 pickle 的能力：
| 能传 | 不能传 |
|---|---|
| 基本类型（int/str/float/bool） | lambda |
| list / dict / set / tuple | 闭包 / 局部函数 |
| 模块级函数 / 类（可 pickle） | 生成器 / 迭代器 |
| 进程 / 线程 / 锁对象 | |
| socket / 文件句柄 |
解决方案：
- 用
dill库：pickle 的增强版，能序列化 lambda、闭包等大部分对象。 - 重构代码：把不可序列化对象改成模块级函数，或把要传的数据改成基本类型（list / dict）再传。
进程同步工具
只讲 Lock（保护共享内存必须用）；其余同步原语与线程章节一致，详见多线程章节。
多进程同样有 Lock / RLock / Semaphore / Event / Condition / Barrier，用法和线程章节几乎一样，只需把 threading 换成 multiprocessing。
唯一在进程里必须掌握的是 Lock —— 保护共享内存的「读改写」操作：
import multiprocessing
def increment(num, lock):
for _ in range(10_000):
with lock: # 保护非原子操作
num.value += 1
if __name__ == "__main__":
lock = multiprocessing.Lock()
num = multiprocessing.Value("i", 0)
ps = [multiprocessing.Process(target=increment, args=(num, lock)) for _ in range(4)]
for p in ps:
p.start()
for p in ps:
p.join()
print(num.value) # 40000其余同步原语（Semaphore 限流、Event 通知、Condition 生产消费、Barrier 栅栏）与线程章节用法一致，不再重复：详见 [[Python 多线程编程]]。
进程池
手写进程需要管理 start / join / 生命周期，代码冗余且容易出错。
进程池自动管理进程的创建、复用和回收，是生产环境的标准姿势。
multiprocessing.Pool
multiprocessing.Pool 是传统的进程池 API：
| 方法 | 返回 | 说明 |
|---|---|---|
map(func, iterable) | list | 批量，按顺序 |
starmap(func, iterable) | list | 批量，参数自动解包 |
apply_async(func, args) | AsyncResult | 单个异步任务 |
imap(func, iterable) | 迭代器 | 惰性，按顺序 |
imap_unordered(func, iterable) | 迭代器 | 惰性，完成顺序 |
from multiprocessing import Pool
def square(x):
return x * x
if __name__ == "__main__":
with Pool(4) as pool: # 4 个进程
print(pool.map(square, range(10))) # [0, 1, 4, 9, ...]
print(pool.starmap(pow, [(2, 3), (3, 2)])) # [8, 9]，自动解包参数
res = pool.apply_async(square, (5,)) # 单个异步
print(res.get()) # 25
for r in pool.imap(square, range(10)): # 惰性、按顺序
print(r, end=" ")
print()
for r in pool.imap_unordered(square, range(10)): # 惰性、完成顺序
print(r, end=" ")
print()ProcessPoolExecutor
concurrent.futures.ProcessPoolExecutor 提供和 ThreadPoolExecutor 完全一致的接口，只需要把线程池换成进程池：
from concurrent.futures import ProcessPoolExecutor
def square(x):
return x * x
if __name__ == "__main__":
with ProcessPoolExecutor(4) as pool:
# submit —— 返回 Future，用 .result() 取值
futures = [pool.submit(square, i) for i in range(10)]
for f in futures:
print(f.result())
# map —— 直接返回结果迭代器
for r in pool.map(square, range(10)):
print(r)两种进程池对比与选择（表格）
multiprocessing.Pool | ProcessPoolExecutor | |
|---|---|---|
| 返回值 | AsyncResult / list | Future |
| 惰性迭代 | imap / imap_unordered | 无（map 一次性取回） |
| 异步提交 | apply_async | submit |
| 异常 | 子进程内手动捕获 | Future.exception() |
| 统一接口 | 独立 API | 与线程池一致，可无痛切换 |
| 适合 | 传统 multiprocessing 场景 | 想在线程/进程池间切换 |
怎么选
代码想在「多进程」和「多线程」之间切换对比，用ProcessPoolExecutor
（换一行 import 即可）；需要惰性迭代处理超大任务，用multiprocessing.Pool
的imap系列。
等待机制：结果什么时候到
submit 拿到 Future 后，取结果有三种方式，差别在「顺序 vs 及时性」：
| 方式 | 结果顺序 | 一句话 | 适用场景 |
|---|---|---|---|
遍历 .result() | 按提交顺序 | 逐个阻塞等 | 任务少、不在乎等待 |
as_completed() | 按完成顺序 | 先完成先处理 | 大量任务，想尽快消费 |
wait() | 不取结果 | 只等一个里程碑 | 阶段同步 |
from concurrent.futures import ProcessPoolExecutor, as_completed
import time
def task(num):
time.sleep(3 - num % 3) # 不同任务耗时不同
return f"任务{num}"
if __name__ == "__main__":
with ProcessPoolExecutor(4) as pool:
futures = [pool.submit(task, i) for i in range(6)]
for f in as_completed(futures): # 谁先完成先输出谁
print(f.result())wait() 不取结果，只等一组任务达到指定状态：
| 条件 | 场景 | 思路 |
|---|---|---|
FIRST_COMPLETED | 竞速请求 | 谁快用谁 |
ALL_COMPLETED | 批量聚合 | 全部拿到再继续 |
FIRST_EXCEPTION | 快速失败 | 有一个错就停 |
与线程章节完全一致
as_completed/wait的用法和坑（按提交顺序干等最慢的任务）都在
多线程章节详细讲过，此处只给结论，细节见 [[Python 多线程编程]]。
异常处理与超时
子进程里的异常会 pickle 回传给父进程，在 Future.result() 时抛出：
from concurrent.futures import ProcessPoolExecutor, TimeoutError
import time
def err_task():
time.sleep(2)
raise ValueError("任务内部报错")
if __name__ == "__main__":
with ProcessPoolExecutor(2) as pool:
f = pool.submit(err_task)
try:
res = f.result(timeout=1)
except TimeoutError:
print("任务执行超时")
except Exception as e:
print("任务异常：", e)常见问题及补充
if name == ‘main’ 必须加（Windows spawn）
Windows 创建子进程用 spawn 方式：不 fork，而是重新启动一个 Python 解释器、重新导入主模块来构造子进程。
没有保护的后果： 子进程重新执行整个脚本 → 又碰到创建进程的代码 → 再创建子进程 → 无限递归 → 程序爆炸。
# ❌ 错误：没有保护，Windows 下无限递归
# import multiprocessing
# def task(): ...
# p = multiprocessing.Process(target=task) # 主进程执行
# p.start() # 子进程又重新执行这两行 → 崩
# ✅ 正确：创建进程的代码放进 if 保护里
import multiprocessing
def task():
print("子进程干活")
if __name__ == "__main__":
p = multiprocessing.Process(target=task)
p.start()
p.join()Linux fork 不受此影响，但为了跨平台一致，建议一律写上。
僵尸进程与孤儿进程
- 僵尸进程（zombie）：子进程已结束，但父进程没有
wait()/join()回收它的退出状态，进程表里留下一个空壳。用join()即可避免。 - 孤儿进程（orphan）：父进程先退出，子进程被系统「收养」，交给 init 进程（PID=1）管理，不会泄漏。
import multiprocessing
import os
import time
def child():
print(f"子进程 {os.getpid()} 开始")
time.sleep(2)
print(f"子进程 {os.getpid()} 结束")
if __name__ == "__main__":
p = multiprocessing.Process(target=child)
p.start()
p.join() # 回收子进程，避免僵尸内存爆炸：每个进程复制一份数据
多进程「独立内存」的另一面：数据被复制多份。8 核机器、每份数据 1GB，理论内存占用可到 8GB。虽然 fork 有写时复制（COW）优化——只在真正修改时才复制，但一旦各进程各自改数据，复制还是会真实发生。
解决方案：
multiprocessing.shared_memory.SharedMemory：真正的共享内存，多进程读同一块字节，不复制。numpy.memmap：把大数组映射到磁盘文件，进程共享同一份，不占内存。
日志混乱：QueueHandler + 独立日志进程
多个进程同时往同一个文件写日志，内容会交错、丢失、甚至损坏文件。
方案：日志进程模式 —— 所有子进程把日志扔进 Queue，由唯一一个日志进程统一负责写文件：
import multiprocessing
import logging
import logging.handlers
import time
def worker(q):
logger = logging.getLogger(f"worker-{multiprocessing.current_process().name}")
logger.setLevel(logging.INFO)
logger.propagate = False # 阻止冒泡到 root 重复输出
logger.addHandler(logging.handlers.QueueHandler(q)) # 发送端：扔进队列
for i in range(3):
logger.info(f"任务 {i}")
time.sleep(0.1)
if __name__ == "__main__":
q = multiprocessing.Queue()
# 接收端：QueueListener 从队列取日志，统一写入文件
file_handler = logging.FileHandler("app.log")
listener = logging.handlers.QueueListener(q, file_handler)
listener.start()
ps = [multiprocessing.Process(target=worker, args=(q,)) for _ in range(4)]
for p in ps:
p.start()
for p in ps:
p.join()
listener.stop()
print("日志已统一写入 app.log")简化结论
记住关键词：QueueHandler（子进程发送）+QueueListener（主进程接收），
就能把多进程日志收敛到同一个文件。
进程数怎么定：os.cpu_count()
import os
print(os.cpu_count()) # 逻辑核数，如 8不是越多越好： 进程有创建、切换、IPC 开销；进程数超过核数后，多出来的进程只能排队等 CPU，反而增加调度开销，可能更慢。
经验法则：
- CPU 密集型：进程数 ≈
os.cpu_count() - I/O 密集型：可以略多于核数（等 I/O 时让位）
- 关键指标：写脚本测不同进程数下的耗时，找「拐点」
实战示例
场景一：CPU 密集型计算对比测试
单进程 vs 多线程 vs 多进程，同样算「区间内质数个数」：
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
def count_primes(n):
cnt = 0
for i in range(2, n):
if all(i % j for j in range(2, int(i ** 0.5) + 1)):
cnt += 1
return cnt
if __name__ == "__main__":
N = 30_000
nums = [N] * 4 # 4 个同样的计算任务
start = time.perf_counter()
seq = [count_primes(n) for n in nums] # 单进程
seq_time = time.perf_counter() - start
start = time.perf_counter()
with ThreadPoolExecutor(4) as pool:
list(pool.map(count_primes, nums)) # 多线程
thread_time = time.perf_counter() - start
start = time.perf_counter()
with ProcessPoolExecutor(4) as pool:
list(pool.map(count_primes, nums)) # 多进程
proc_time = time.perf_counter() - start
print(f"单进程: {seq_time:.2f}s")
print(f"多线程: {thread_time:.2f}s") # 受 GIL 限制，≈ 单进程
print(f"多进程: {proc_time:.2f}s") # 约 单进程/核数场景二：大规模数据并行处理（Pool + pandas 分块）
大数据集按行分块，每块交给一个进程处理，最后合并：
import pandas as pd
from multiprocessing import Pool
def process_chunk(chunk):
"""对每一块做耗时处理（示例：按类别算均值）"""
chunk['mean_by_cat'] = chunk.groupby('category')['value'].transform('mean')
return chunk
if __name__ == "__main__":
df = pd.read_csv("big_data.csv")
n = 4 # 假设 4 核
chunks = [df.iloc[i::n] for i in range(n)] # 按行均分 4 块
with Pool(n) as pool:
parts = pool.map(process_chunk, chunks)
result = pd.concat(parts)
print(result.head())场景三：守护进程与后台任务
启动一个长期运行的后台任务（比如定时心跳上报），主进程干完自己的事直接退出，后台进程用 daemon=True 随主进程结束而回收：
import multiprocessing
import time
def heartbeat():
while True:
print(f"[{time.strftime('%H:%M:%S')}] 心跳...")
time.sleep(1)
def main_work():
time.sleep(3)
print("主任务完成")
if __name__ == "__main__":
bg = multiprocessing.Process(target=heartbeat, daemon=True)
bg.start()
main_work()
# 主进程退出，daemon 后台进程被自动回收想让后台任务不随主进程退出？用
daemon=False，
但此时主进程必须join()或自己设计退出条件，否则程序不会结束。
结语
多进程解决了「算」的问题——CPU 密集型任务真正利用多核，绕过 GIL，把计算速度成倍提高。
但它也有代价：进程独立内存、通信靠序列化、创建开销大。当面对大量 I/O 等待（网络请求、数据库查询）时，CPU 大部分时间在空转，瓶颈不在算力，在等待。这时候用多进程太沉重。
下一章，我们用多线程解决「等」的问题：共享内存、创建轻量、切换高效——在讨论更轻量的 asyncio 之前，先打好这个基础。
参考资料
- multiprocessing 官方文档
- concurrent.futures 官方文档
- shared_memory 官方文档
- [[Python 多线程编程]]（系列下一章，多处正文引用）
- asyncio 官方文档（下一章预读）