Python多进程中如何实现任务动态分配以优化运行时长？

问题 #

我正在解决Python中的一个并行化问题。假设我有需要处理的数据，且使用multiprocessing模块运行2个并行进程。

我可以将数据拆分为两部分，分别分配给两个进程：

data = [a, b, c, d, e]
data_process1 = [a, b, c]
data_process2 = [d, e]

但每个数据点的处理时长各不相同，比如a、b、c各需1分钟，d、e各需10分钟，且处理前无法预估耗时。这种静态分配方式下，总运行时长为20分钟（进程1耗时3分钟，进程2耗时20分钟）。

我的想法是：不预先拆分数据分配给进程，而是将数据存入队列实现动态分配，每当进程空闲时就从队列中获取下一个任务。这样进程1完成任务后可接手e，总运行时长可降至13分钟。请问是否可以实现上述动态任务分配方式？

解决方案 #

完全可以实现这种动态任务分配方式，用multiprocessing.Queue就能完成核心的任务分发逻辑，让空闲进程自动从队列中获取新任务。

实现思路 #

• 创建一个任务队列，将所有待处理数据放入队列
• 定义worker进程逻辑：持续从队列取任务，处理完成后继续取下一个，直到队列清空
• 启动指定数量的进程（此处为2个），让它们共享这个任务队列

代码示例 #

import multiprocessing
import time

# 模拟不同耗时的任务处理函数
def process_task(task):
    if task in ['d', 'e']:
        time.sleep(10)  # 模拟d、e耗时10分钟（用秒代替）
    else:
        time.sleep(1)   # 模拟a、b、c耗时1分钟（用秒代替）
    print(f"完成任务: {task}")

# worker进程逻辑：从队列取任务并处理
def worker(task_queue):
    while not task_queue.empty():
        try:
            task = task_queue.get(block=False)
            process_task(task)
            task_queue.task_done()
        except multiprocessing.queues.Empty:
            break

if __name__ == "__main__":
    start_time = time.time()
    
    # 初始化任务队列并填充数据
    task_queue = multiprocessing.JoinableQueue()
    data = ['a', 'b', 'c', 'd', 'e']
    for task in data:
        task_queue.put(task)
    
    # 创建并启动2个进程
    processes = []
    for _ in range(2):
        p = multiprocessing.Process(target=worker, args=(task_queue,))
        processes.append(p)
        p.start()
    
    # 等待所有任务完成
    task_queue.join()
    
    # 等待进程结束
    for p in processes:
        p.join()
    
    end_time = time.time()
    print(f"总耗时: {end_time - start_time:.2f} 秒")

效果说明 #

运行代码后总耗时会接近13秒（对应你所说的13分钟）：

• 进程1先处理a、b、c（共3秒），空闲后立即接手处理e（10秒）
• 进程2全程处理d（10秒），完成后等待队列任务结束
• 总耗时为3+10=13秒左右，完全实现了动态任务调度的优化效果

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