Python 进程池 ProcessPoolExecutor 使用指南

一、进程池概述

进程池（ProcessPoolExecutor）是 Python 中用于并行执行任务的强大工具，尤其适合 CPU 密集型操作。与传统的多进程编程相比，它提供了更简单、更高级的接口。

适用场景

1. CPU 密集型任务（数学计算、图像处理等）
2. 需要并行处理独立任务的情况
3. 需要限制并发进程数量的场景
4. 需要获取任务执行结果的场景

---

二、基本使用

from concurrent.futures import ProcessPoolExecutor
import time

# CPU 密集型计算函数
def calculate_square(n):
    print(f"计算 {n} 的平方...")
    time.sleep(1)  # 模拟耗时计算
    return n * n

# 使用进程池
with ProcessPoolExecutor(max_workers=4) as executor:
    # 提交任务到进程池
    future1 = executor.submit(calculate_square, 5)
    future2 = executor.submit(calculate_square, 8)
    
    # 获取任务结果
    print(f"5 的平方 = {future1.result()}")
    print(f"8 的平方 = {future2.result()}")

---

三、核心方法详解

1. 任务提交

map(): 批量提交任务

results = executor.map(func, iterable, timeout=None)

submit(): 提交单个任务

future = executor.submit(func, *args, **kwargs)

2. 结果处理

as_completed(): 按照完成顺序获取结果

from concurrent.futures import as_completed

futures = [executor.submit(calculate_square, i) for i in range(1, 6)]
for future in as_completed(futures):
    print(f"结果：{future.result()}")

future.result(timeout=None): 获取任务结果（阻塞）

result = future.result()  # 阻塞直到结果返回

---

四、高级用法

1. 限制并发进程数

# 最多同时运行 2 个进程
with ProcessPoolExecutor(max_workers=2) as executor:
    results = list(executor.map(calculate_square, range(1, 5)))
    print(results)

2. 获取任务状态

future = executor.submit(calculate_square, 10)
if future.running():
    print("任务正在运行...")
elif future.done():
    print("任务已完成!")

3. 回调处理结果

def result_callback(future):
    print(f"收到结果：{future.result()}")

with ProcessPoolExecutor() as executor:
    future = executor.submit(calculate_square, 15)
    future.add_done_callback(result_callback)

4. 处理异常

def divide(a, b):
    return a / b

try:
    future = executor.submit(divide, 10, 0)
    result = future.result()
except ZeroDivisionError as e:
    print(f"出现错误：{e}")

---

五、实际应用案例

案例：批量图片处理

from PIL import Image
import os
from concurrent.futures import ProcessPoolExecutor

# 图片处理函数
def process_image(image_path):
    try:
        img = Image.open(image_path)
        # 图片处理操作
        img = img.resize((800, 600))
        img = img.convert('L')  # 转为灰度图
        # 保存处理后的图片
        new_path = os.path.splitext(image_path)[0] + "_processed.jpg"
        img.save(new_path)
        return f"已处理：{image_path}"
    except Exception as e:
        return f"处理失败：{image_path} - {str(e)}"

# 获取图片目录中的所有图片
image_dir = "images"
image_files = [
    os.path.join(image_dir, f)
    for f in os.listdir(image_dir)
    if f.endswith(('.jpg', '.png'))
]

# 使用进程池处理
with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
    # 提交所有任务
    futures = {
        executor.submit(process_image, img): img 
        for img in image_files
    }
    # 获取结果
    for future in as_completed(futures):
        result = future.result()
        print(result)

---

六、性能优化技巧

1. 选择合适的 max_workers:
   • 对于 CPU 密集型任务：max_workers=os.cpu_count()
   • 对于 I/O 密集型任务：max_workers=(os.cpu_count() * 2)
2. 减少数据传输:
   • 避免在进程间传递大对象
   • 使用共享内存 (SharedMemory) 或服务器进程 (Manager) 优化数据共享

任务分块:

# 减少小任务的数量
def process_chunk(chunk):
    return [calculate_square(n) for n in chunk]

chunks = [range(i, i+1000) for i in range(0, 10000, 1000)]
results = executor.map(process_chunk, chunks)

预加载数据:

# 使用 initializer 预加载共享数据
def init_worker():
    global shared_data
    shared_data = load_big_data()

def process_item(item):
    return process(shared_data, item)

with ProcessPoolExecutor(initializer=init_worker) as executor:
    ...

---

七、常见问题解决方案

问题 1：子进程异常导致无限等待

解决方案:

# 设置超时时间
try:
    result = future.result(timeout=60)  # 最多等待 60 秒
except TimeoutError:
    print("任务超时")

问题 2：子进程不被回收

解决方案:

# 使用上下文管理器确保资源回收
with ProcessPoolExecutor() as executor:
    # 执行代码
    # 离开 with 块后自动关闭进程池

问题 3：共享数据问题

解决方案:

from multiprocessing import Manager

def worker(shared_list, data):
    shared_list.append(process(data))

with Manager() as manager:
    shared_list = manager.list()
    with ProcessPoolExecutor() as executor:
        executor.map(worker, [shared_list]*len(data), data)
    print(list(shared_list))

---

八、与线程池的选择建议

特性	进程池 (ProcessPoolExecutor)	线程池 (ThreadPoolExecutor)
适用任务	CPU 密集型	I/O 密集型
内存使用	高 (每个进程独立内存空间)	低 (共享内存)
上下文切换开销	高	低
GIL 限制	避免 GIL 影响	受 GIL 限制
数据共享	复杂 (需要专门机制)	简单 (直接共享)
通信开销	高 (需要序列化)	低 (直接内存访问)

选择建议:

• 优先考虑线程池处理 I/O 密集型任务
• 仅当任务受 GIL 限制时使用进程池
• 混合使用：I/O 密集型任务使用线程池，CPU 密集型任务使用进程池

---

九、结语

ProcessPoolExecutor 是 Python 并发编程的核心组件之一，熟练掌握它可以显著提升程序性能。关键要点：

1. 使用上下文管理器 (with语句) 确保资源正确释放
2. 根据任务类型选择合理的 max_workers 数量
3. 优先使用 map() 和 as_completed() 管理批量任务
4. 处理好任务间的数据共享问题
5. 针对不同任务特点优化参数配置