一、为什么需要进程池？

Python 的 GIL（全局解释器锁）使得同一时刻只有一个线程能执行 Python 字节码，这意味着多线程在 CPU 密集型任务上几乎无法获得真正的并行加速。要绕过 GIL，就必须使用多进程。

但如果为每个任务都手动创建、启动、销毁一个进程，开销会非常大——进程的创建比线程重得多，涉及内存复制和操作系统调度。进程池正是为此而生：它预先创建固定数量的工作进程，重复利用它们来处理一批又一批任务，从而把创建开销摊薄。

Python 标准库提供了两种进程池实现：

• concurrent.futures.ProcessPoolExecutor（高层接口，推荐）
• multiprocessing.pool.Pool（传统接口，功能更丰富）

本文以 ProcessPoolExecutor 为主线，逐步拆解它的核心概念：Future、as_completed、add_done_callback 以及 map，最后再介绍传统 Pool 的对应方法及两者选型建议。

---

二、快速上手：ProcessPoolExecutor

2.1 创建进程池并提交任务

from concurrent.futures import ProcessPoolExecutor
import os

def cpu_bound_task(n):
    """模拟一个 CPU 密集型计算"""
    result = sum(i * i for i in range(10**6))
    return f"进程{os.getpid()} 处理参数 {n}，结果={result}"

# 使用 with 语句自动管理池的生命周期
with ProcessPoolExecutor(max_workers=4) as executor:
    # submit：提交单个任务，立即返回一个 Future 对象
    future = executor.submit(cpu_bound_task, 42)
    
    # 通过 future.result() 阻塞等待结果
    result = future.result()
    print(result)

max_workers 控制工作进程数量，通常设置为 os.cpu_count()。缺省时 ProcessPoolExecutor 会自动取 CPU 核心数。

2.2 深入理解 Future 对象

executor.submit() 不会直接返回计算结果，而是返回一个 Future 对象。可以把 Future 理解为一个“未来的占位符”——它在任务提交时就拿到，任务完成后再通过它取结果。

Future 的关键方法和属性：

方法 / 属性	说明
.result(timeout=None)	阻塞等待，返回任务结果；超时抛出 TimeoutError
.exception(timeout=None)	返回任务抛出的异常对象（不重新抛出），无异常则返回 None
.done()	检查任务是否已完成（成功或失败都算）
.running()	检查任务是否正在执行
.cancelled()	检查任务是否被取消
.cancel()	尝试取消任务（对已运行的任务通常无效）
.add_done_callback(fn)	添加完成回调函数

典型的错误处理模式：

from concurrent.futures import ProcessPoolExecutor

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

with ProcessPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(safe_divide, 10, i) for i in [2, 0, 5]]
    
    for future in futures:
        exc = future.exception()
        if exc is not None:
            print(f"任务失败：{type(exc).__name__}: {exc}")
        else:
            print(f"结果：{future.result()}")

输出：

结果：5.0
任务失败：ZeroDivisionError: division by zero
结果：2.0

future.exception() 的妙处在于不重新抛出异常，让你可以优雅地逐个处理错误。

---

三、批量提交：map 方法

如果有一批相同的函数要应用到一系列参数上，用 submit 逐个提交太繁琐。executor.map 可以一次提交所有任务，并按参数顺序返回结果：

from concurrent.futures import ProcessPoolExecutor

def square(n):
    return n * n

with ProcessPoolExecutor(max_workers=4) as executor:
    # map 按参数顺序返回结果
    results = executor.map(square, [1, 2, 3, 4, 5])
    print(list(results))  # [1, 4, 9, 16, 25]

map 的签名是 map(fn, *iterables, timeout=None, chunksize=1)：

• *iterables：支持多个可迭代对象，类似内置 map
• timeout：整体超时时间
• chunksize：将输入切分成块分配给进程，对性能有直接影响

多参数任务 可以这样传：

def add(a, b):
    return a + b

with ProcessPoolExecutor() as executor:
    results = executor.map(add, [1, 2, 3], [10, 20, 30])
    print(list(results))  # [11, 22, 33]

chunksize 的调优价值：当任务非常短小（比如几千个简单计算），进程间通信开销会变得显著。增大 chunksize 让每个进程一次拿一批任务，减少通信轮次，可显著提升性能。反之，如果各任务耗时差异很大，chunksize=1 能更好地负载均衡。

注意：executor.map 是顺序产出结果的。即使后面的任务先完成，也得等前面的结果就绪才能继续迭代。如果需要“谁先完成就先处理谁”，就要用到下一节的 as_completed。

---

四、按完成顺序处理：as_completed

这是本文的重点之一。concurrent.futures.as_completed 接收一组 Future，返回一个迭代器，哪个 Future 先完成，就先产出哪个。

from concurrent.futures import ProcessPoolExecutor, as_completed
import time, random

def delayed_square(n):
    time.sleep(random.uniform(0.3, 1.5))  # 模拟耗时波动
    return n, n * n

with ProcessPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(delayed_square, i) for i in range(8)]
    
    for future in as_completed(futures):
        n, sq = future.result()
        print(f"参数 {n} → 结果 {sq}")

输出会按实际完成顺序排列，而非参数顺序。这对于可以流式处理结果的场景非常有用——你不需要等最慢的任务完成就能开始处理已完成的部分。

as_completed 还支持 timeout 参数：如果等待超过指定秒数仍无任何任务完成，抛出 TimeoutError。

和 map 的对比：

	executor.map	as_completed
产出顺序	按参数顺序	按完成顺序
内存占用	低（惰性迭代）	需要先生成所有 Future 再传入
适合场景	结果需保持对应关系	流式处理、尽快消费结果

---

五、异步回调：add_done_callback

除了主动轮询 Future 的结果，还可以用 add_done_callback 注册一个回调函数，任务完成后自动调用：

from concurrent.futures import ProcessPoolExecutor

def on_task_done(future):
    """回调函数接收 Future 对象作为参数"""
    try:
        result = future.result()
        print(f"回调：任务完成，结果={result}")
    except Exception as e:
        print(f"回调：任务失败，{e}")

with ProcessPoolExecutor(max_workers=4) as executor:
    for i in range(5):
        future = executor.submit(pow, i, 2)
        future.add_done_callback(on_task_done)
    
    # 回调在 Future 设置结果时触发

关键细节：

• 回调函数的执行线程/进程不固定——对于 ProcessPoolExecutor，回调在设置结果的那个线程中执行（通常是主线程或某个内部线程），不是在子进程里。
• 回调执行顺序不一定等于任务完成顺序，取决于调度。
• 回调中的异常不会被传播，必须在回调内部自行捕获。
• 如果有多个回调注册到同一个 Future，它们按注册顺序依次执行。

回调模式适合触发后续操作的场景，比如“任务完成后写入日志”、“将结果放入队列”、“更新进度条”等。

---

六、另一种选择：multiprocessing.pool.Pool

ProcessPoolExecutor 虽然是推荐方案，但传统 multiprocessing.pool.Pool 有其独特的 API 设计，了解它有助于阅读旧代码或使用某些特定功能。

from multiprocessing import Pool

def cube(n):
    return n ** 3

with Pool(processes=4) as pool:
    # 同步 map，阻塞直到全部完成
    print(pool.map(cube, range(5)))  # [0, 1, 8, 27, 64]
    
    # 异步 map_async，返回 AsyncResult
    async_res = pool.map_async(cube, range(5))
    print(async_res.get())  # [0, 1, 8, 27, 64]
    
    # 按完成顺序产出（对应 as_completed）
    for result in pool.imap_unordered(cube, range(5)):
        print(f"完成：{result}")

两者的概念对应关系：

ProcessPoolExecutor	multiprocessing.pool.Pool
submit(fn, *args)	apply_async(fn, args, ...)
map(fn, iterable)	map(fn, iterable)
as_completed(futures)	imap_unordered(fn, iterable)
Future 对象	AsyncResult 对象
future.add_done_callback(fn)	apply_async(callback=fn)
没有 starmap	starmap 支持多参元组展开

选型建议：新项目优先用 ProcessPoolExecutor，API 更简洁、与线程池接口统一、和 asyncio 协作更友好。如果要用 starmap 或需要更细粒度的异步控制，再考虑 Pool。

---

七、最佳实践与常见陷阱

7.1 序列化限制

进程间不共享内存，所有参数和返回值都必须通过 pickle 序列化传递。不能传递 lambda、带闭包的函数、某些特殊对象：

# ❌ 错误：lambda 无法 pickle
executor.submit(lambda x: x + 1, 5)  # PicklingError

解决方案：把函数定义在模块顶层，或使用 multiprocessing 的 Manager 共享代理对象。

7.2 启动方式与主模块保护

Windows 和 macOS 默认使用 spawn 启动方式（Linux 用 fork）。在 spawn 下，子进程会重新导入主模块，因此必须用 if __name__ == '__main__': 保护入口代码，否则会无限递归创建进程：

# ✅ 正确写法
if __name__ == '__main__':
    with ProcessPoolExecutor() as executor:
        results = executor.map(do_work, data)

7.3 使用 initializer 初始化每个进程

如果每个子进程需要独立的资源（如数据库连接、日志配置），用 initializer 参数：

import logging
from concurrent.futures import ProcessPoolExecutor

def init_worker():
    logging.basicConfig(level=logging.INFO)
    print(f"进程初始化完成")

with ProcessPoolExecutor(max_workers=4, initializer=init_worker) as executor:
    ...

7.4 合理设置进程数

• CPU 密集型：max_workers = os.cpu_count()
• 混合型任务：可以略多于 CPU 数量，让 I/O 等待时也有进程可用
• 避免设置过大，超多进程会引发上下文切换开销和内存压力

7.5 大数据传参的优化

如果任务需要处理大型数据集，不要通过参数传递整个数据，而是传索引或文件名，让子进程自己去加载：

# ❌ 低效：每个子进程都序列化整个大数组
big_data = load_huge_array()
executor.map(process_chunk, [big_data[i::4] for i in range(4)])

# ✅ 高效：只传索引，子进程按需加载
def process_by_index(idx):
    data = load_huge_array()  # 每个进程独立加载
    return process(data[idx::4])

executor.map(process_by_index, range(4))

7.6 避免在任务内部再次提交到同一池

这容易导致死锁——所有工作进程都在等待彼此的任务完成，而池中没有空闲进程来执行新任务。

---

八、实战案例：批量图片缩略图生成

下面是一个完整的实战示例，综合运用了 submit、as_completed 和回调：

import os, time, random
from concurrent.futures import ProcessPoolExecutor, as_completed
from PIL import Image  # 需要 pip install Pillow

THUMB_SIZE = (128, 128)

def create_thumbnail(image_path, output_dir):
    """生成缩略图并返回处理信息"""
    time.sleep(random.uniform(0.1, 0.5))  # 模拟耗时
    img = Image.open(image_path)
    img.thumbnail(THUMB_SIZE)
    
    filename = os.path.basename(image_path)
    out_path = os.path.join(output_dir, f"thumb_{filename}")
    img.save(out_path)
    return image_path, out_path

def log_completion(future):
    """完成回调：记录日志"""
    try:
        src, dst = future.result()
        print(f"✅ {src} → {dst}")
    except Exception as e:
        print(f"❌ 处理失败: {e}")

if __name__ == '__main__':
    image_dir = "./images"
    output_dir = "./thumbnails"
    os.makedirs(output_dir, exist_ok=True)
    
    images = [
        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 = []
        for img in images:
            f = executor.submit(create_thumbnail, img, output_dir)
            f.add_done_callback(log_completion)
            futures.append(f)
        
        # 按完成顺序实时输出（as_completed 保证先完成的先被处理）
        for future in as_completed(futures):
            # 这里可以叠加额外的实时处理逻辑
            pass  # 回调已经处理了日志，这里仅做同步等待

    print(f"\n全部完成！共处理 {len(images)} 张图片")

这个案例中：

• submit 逐个提交图片处理任务
• add_done_callback 注册日志回调，完成即输出
• as_completed 确保主流程等待所有任务完成

---

九、总结

概念	一句话总结
进程池	复用一组进程执行 CPU 密集型任务，绕过 GIL，实现真正并行
Future	任务提交后立即返回的“占位符”，通过 .result() / .exception() 获取结果
map	批量提交，按参数顺序返回结果，内存友好
as_completed	按完成顺序产出结果，适合流式处理
add_done_callback	注册回调函数，任务完成后自动触发，适合日志、通知等副作用操作

掌握这些，你就能在 Python 中高效地驾驭多进程并行计算。根据不同场景灵活选择 map、as_completed 或回调模式，就能在保持代码简洁的同时榨干多核 CPU 的性能。