Python multiprocessing.Pool并行任务：分块大小选择与中断处理

一、分块大小（chunksize）的选择策略 #

multiprocessing.Pool的chunksize参数直接影响进程间通信开销与负载均衡：

• 太小：进程频繁接收小任务块，IPC开销占比高，拖慢整体速度
• 太大：部分进程提前完成任务后闲置，导致负载不均衡

针对你的Intel i9（8核4逻辑）设备，以及数千张图片的场景，按以下步骤选择最优值：

1. 计算基准值：假设总任务数为total_tasks，进程数设为processes=8（物理核数对CPU密集型任务的利用率更稳定），基准chunksize为total_tasks // processes。比如1000张图，基准值是125。
2. 小范围调优测试：在基准值的±50%范围内选3-5个值（比如75、125、175），用相同的图片子集（比如200张）跑测试，记录总耗时。
3. 考虑任务波动：如果图片大小差异大（单任务耗时波动超过20%），选稍小的chunksize；如果图片规格接近，选稍大的chunksize。
4. 固化最优值：将测试出的最优chunksize应用到全量任务，后续同类型任务可直接复用。

测试代码示例：

import time
from multiprocessing import Pool
import subprocess

def scale_icon(img_path):
    # 调用magick的核心逻辑
    try:
        subprocess.run(["magick", img_path, "-resize", "64x64", f"resized_{img_path}"], check=True, capture_output=True)
    except subprocess.CalledProcessError:
        return False
    return True

def test_chunksize(chunksize, img_list):
    start = time.time()
    with Pool(processes=8) as pool:
        pool.map(scale_icon, img_list, chunksize=chunksize)
    end = time.time()
    print(f"Chunksize {chunksize}: 耗时 {end - start:.2f}s")

if __name__ == "__main__":
    # 用200张图做测试子集
    test_imgs = [f"img_{i}.png" for i in range(200)]
    for cs in [75, 125, 175]:
        test_chunksize(cs, test_imgs)

二、两种中断需求的实现方案 #

需求1：遇错后完成队列中所有任务再停止 #

核心思路：用共享变量标记错误状态，让所有已提交任务执行完毕，之后终止Pool不再处理新任务。

代码示例：

from multiprocessing import Pool, Value
import subprocess
import ctypes

# 共享错误标志：0=正常，1=已触发错误
error_occurred = Value(ctypes.c_int, 0)

def scale_icon(img_path):
    # 可选：若已触发错误，直接跳过当前任务
    if error_occurred.value == 1:
        return (img_path, "skipped due to prior error")
    try:
        subprocess.run(["magick", img_path, "-resize", "64x64", f"resized_{img_path}"], check=True, capture_output=True)
        return (img_path, "success")
    except subprocess.CalledProcessError as e:
        # 原子性设置错误标志，避免多进程重复设置
        with error_occurred.get_lock():
            if error_occurred.value == 0:
                error_occurred.value = 1
        return (img_path, f"failed: {str(e)}")

if __name__ == "__main__":
    img_list = [f"img_{i}.png" for i in range(1000)]
    with Pool(processes=8) as pool:
        results = pool.map(scale_icon, img_list, chunksize=125)
    
    # 统计错误结果
    failed = [res for res in results if "failed" in res[1]]
    if failed:
        print(f"检测到{len(failed)}个错误，已完成所有队列任务")

需求2：遇错后立即终止所有任务 #

核心思路：用multiprocessing.Event触发中断，主进程监控事件，一旦触发就调用pool.terminate()强制终止所有子进程。

代码示例：

from multiprocessing import Pool, Event
import subprocess
import time

# 中断事件
interrupt_event = Event()

def scale_icon(img_path):
    if interrupt_event.is_set():
        return (img_path, "interrupted")
    try:
        subprocess.run(["magick", img_path, "-resize", "64x64", f"resized_{img_path}"], check=True, capture_output=True)
        return (img_path, "success")
    except subprocess.CalledProcessError as e:
        # 设置中断事件，触发主进程终止逻辑
        interrupt_event.set()
        return (img_path, f"failed: {str(e)}")

def monitor_interrupt(pool):
    # 主进程监控事件，触发则终止Pool
    while not interrupt_event.is_set():
        time.sleep(0.5)
    pool.terminate()
    print("已触发立即中断，所有任务终止")

if __name__ == "__main__":
    img_list = [f"img_{i}.png" for i in range(1000)]
    with Pool(processes=8) as pool:
        # 启动监控线程
        import threading
        monitor_thread = threading.Thread(target=monitor_interrupt, args=(pool,))
        monitor_thread.start()
        
        try:
            results = pool.map(scale_icon, img_list, chunksize=125)
        except Exception:
            # 捕获Pool终止后的异常
            pass
        
        monitor_thread.join()

注意事项：

• 使用terminate()后，子进程会被强制杀死，可能导致临时文件残留，需在代码中增加清理逻辑。
• 共享变量/事件必须通过multiprocessing提供的类创建，避免普通全局变量无法跨进程同步的问题。

内容的提问来源于stack exchange，提问作者Prashant