Python如何实现固定并行数任务：完成一个自动补新任务

核心问题说明 #

原生multiprocessing.Process没有提供「等待任意一个进程结束」的内置API，且直接调用Process时无法获取target函数的返回值，需要借助进程间通信机制或者高层封装实现需求。

方案1：基于原生multiprocessing实现（适配现有伪代码结构） #

通过multiprocessing.Queue传递任务结果，利用队列的阻塞等待特性实现任意进程完成的监听，同时补位新进程维持并行数。

import time
import random
from multiprocessing import Process, Queue

# 总任务数（根据实际业务需求调整，所有任务跑完后自动退出）
TOTAL_TASKS = 10

def foo(task_id, result_queue):
    """测试任务：随机休眠2~10秒，执行完后将结果写入队列"""
    run_time = random.randint(2, 10)
    time.sleep(run_time)
    # 写入队列标识任务完成，传递任务结果
    result_queue.put((task_id, 42, run_time))

if __name__ == "__main__":
    active_processes = []  # 存储运行中的进程及对应任务ID
    result_queue = Queue()  # 进程间通信的结果队列
    next_task_id = 0

    # 初始启动3个进程
    for _ in range(3):
        p = Process(target=foo, args=(next_task_id, result_queue))
        p.start()
        active_processes.append((p, next_task_id))
        next_task_id += 1

    completed_count = 0
    while completed_count < TOTAL_TASKS:
        # 阻塞等待任意进程写入完成消息（等价于等待任意进程结束）
        task_id, result, run_time = result_queue.get()
        completed_count += 1
        print(f"任务{task_id}完成，结果：{result}，耗时：{run_time}s")

        # 从活跃列表移除已结束的进程
        active_processes = [(p, tid) for p, tid in active_processes if tid != task_id]

        # 还有未执行的任务则启动新进程补位，维持3个并行
        if next_task_id < TOTAL_TASKS:
            new_p = Process(target=foo, args=(next_task_id, result_queue))
            new_p.start()
            active_processes.append((new_p, next_task_id))
            next_task_id += 1

    # 回收进程资源
    for p, _ in active_processes:
        p.join()
    print("所有任务执行完毕")

方案2：使用concurrent.futures高层封装（更简洁稳定） #

Python 3.2+内置的concurrent.futures.ProcessPoolExecutor已内置固定并行数调度、进程回收、返回值捕获能力，不需要手动维护进程列表和等待逻辑，更适合生产环境使用。

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

# 总任务数（根据实际业务需求调整）
TOTAL_TASKS = 10

def foo(task_id):
    """测试任务：随机休眠2~10秒，直接返回结果"""
    run_time = random.randint(2, 10)
    time.sleep(run_time)
    return task_id, 42, run_time

if __name__ == "__main__":
    # 初始化最多3个并行进程的执行器，自动维持并行数
    with ProcessPoolExecutor(max_workers=3) as executor:
        # 提交所有任务，执行器会自动调度：始终保持最多3个进程运行，进程结束自动领取新任务
        futures = [executor.submit(foo, task_id) for task_id in range(TOTAL_TASKS)]

        # 按完成顺序获取结果，任意任务完成时立刻触发迭代
        for future in as_completed(futures):
            task_id, result, run_time = future.result()
            print(f"任务{task_id}完成，结果：{result}，耗时：{run_time}s")

    print("所有任务执行完毕")

适配场景说明 #

• 动态任务场景：如果任务不是提前全部生成，而是运行过程中动态产生，可在循环中按需调用executor.submit(new_foo_task)提交新任务，执行器会自动调度补位，始终保持3个并行进程。
• IO密集型任务：可将ProcessPoolExecutor替换为ThreadPoolExecutor，使用方式完全一致，无需修改其他逻辑。

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