如何在Python multiprocessing中限制并发worker的最大数量？

嘿，完全懂你的需求——你不想等所有串行的heavy_lifting都做完再批量并行处理，而是想处理完一个就立刻把可并行的任务扔出去，同时还要把并发进程数牢牢控制在X个，对吧？

其实multiprocessing.Pool完全能满足你的需求，关键是用非阻塞的apply_async方法，而不是你担心的map（后者确实要等所有任务都提交后才会开始处理）。下面我给你改个例子，贴合你的循环场景：

import multiprocessing

def heavy_lifting_which_cannot_be_parallelized(item):
    # 这里是你的串行计算逻辑，比如处理单个item的耗时操作
    print(f"正在串行处理item: {item}")
    return f"processed_{item}"

def process_result_in_a_way_that_can_be_parallelized(result, return_dict, item_id):
    # 这里是可并行的处理逻辑
    print(f"并行处理结果: {result}")
    # 把结果存到线程安全的字典里
    return_dict[item_id] = f"final_{result}"

if __name__ == '__main__':
    # 设置你要限制的并发进程数X
    MAX_CONCURRENT_PROCESSES = 2
    items = [1, 2, 3, 4, 5]
    
    # 创建进程池，限制并发数
    pool = multiprocessing.Pool(processes=MAX_CONCURRENT_PROCESSES)
    # 用Manager创建线程安全的字典来收集结果（主进程和子进程都能访问）
    manager = multiprocessing.Manager()
    return_dict = manager.dict()
    
    for idx, item in enumerate(items):
        # 先执行串行的heavy lifting
        result = heavy_lifting_which_cannot_be_parallelized(item)
        # 异步提交并行任务，非阻塞，提交后立刻回到循环处理下一个item
        pool.apply_async(
            func=process_result_in_a_way_that_can_be_parallelized,
            args=(result, return_dict, idx)
        )
    
    # 关闭进程池，不再接受新任务
    pool.close()
    # 等待所有提交的任务执行完成
    pool.join()
    
    # 输出最终结果
    print("所有任务完成，结果如下:")
    print(sorted(return_dict.items()))

关键细节说明： #

• Pool(processes=X)：直接通过processes参数设置最大并发进程数，Pool会自动帮你管理进程的创建、复用和排队，不会超过X个进程同时运行。
• apply_async：这个方法是非阻塞的——提交任务后，主进程会立刻回到循环继续处理下一个item的串行逻辑，而不是等待当前并行任务完成。当Pool里有空闲进程时，就会自动取出排队的任务执行。
• 线程安全的结果收集：用Manager().dict()来存结果，因为子进程不能直接修改主进程的普通字典，Manager提供的容器是跨进程安全的。如果你不需要收集结果，这部分可以省略。
• close()和join()：必须调用close()告诉Pool不再接受新任务，然后join()等待所有已提交的任务全部执行完毕，避免主进程提前退出。

这样一来，你就能一边做串行的计算，一边异步提交并行任务，同时严格控制并发数，完美贴合你的需求～

内容的提问来源于stack exchange，提问作者The Unfun Cat