Python多进程任务调度与计时问题技术求助

一、使用Process的正确实现 #

你的核心问题是未等待所有进程执行完毕就计算总耗时，同时需要严格控制任务的依赖顺序。以下是适配需求的代码：

import multiprocessing
import time

# 定义任务函数，模拟耗时操作
def job(name, delay):
    print(f"启动任务 {name}")
    time.sleep(delay)
    print(f"完成任务 {name}")

if __name__ == "__main__":
    start_time = time.time()
    # 限制最大并发进程数，适配4核环境
    # 此处最多同时运行3个任务，符合4核资源限制
    # 启动job1、job2、job3
    p1 = multiprocessing.Process(target=job, args=("job1", 2))
    p2 = multiprocessing.Process(target=job, args=("job2", 3))
    p3 = multiprocessing.Process(target=job, args=("job3", 1))
    
    p1.start()
    p2.start()
    p3.start()
    
    # 等待job3完成后启动job4
    p3.join()
    p4 = multiprocessing.Process(target=job, args=("job4", 2))
    p4.start()
    
    # 等待job4完成后启动job5
    p4.join()
    p5 = multiprocessing.Process(target=job, args=("job5", 1))
    p5.start()
    
    # 等待job5完成后启动job6
    p5.join()
    p6 = multiprocessing.Process(target=job, args=("job6", 2))
    p6.start()
    
    # 等待所有剩余进程完成
    p1.join()
    p2.join()
    p6.join()
    
    total_time = time.time() - start_time
    print(f"所有任务完成，总耗时: {total_time:.2f}秒")

关键说明： #

• 必须在所有进程调用join()后再计算总耗时，确保主线程等待所有子进程结束。
• 通过控制并发进程数不超过4，避免4核机器资源耗尽导致卡顿。
• 利用join()严格控制任务依赖顺序：job3→job4→job5→job6。

二、使用Pool的正确实现 #

map_async不适合处理有依赖顺序的任务，改用apply_async提交单个任务，并通过get()等待依赖任务完成。以下是实现代码：

import multiprocessing
import time

def job(name, delay):
    print(f"启动任务 {name}")
    time.sleep(delay)
    print(f"完成任务 {name}")
    return f"{name} finished"

if __name__ == "__main__":
    start_time = time.time()
    # 初始化进程池，大小设为4，适配4核环境
    with multiprocessing.Pool(processes=4) as pool:
        # 提交job1、job2、job3，异步执行
        res1 = pool.apply_async(job, args=("job1", 2))
        res2 = pool.apply_async(job, args=("job2", 3))
        res3 = pool.apply_async(job, args=("job3", 1))
        
        # 等待job3完成，再提交job4
        res3.get()
        res4 = pool.apply_async(job, args=("job4", 2))
        
        # 等待job4完成，再提交job5
        res4.get()
        res5 = pool.apply_async(job, args=("job5", 1))
        
        # 等待job5完成，再提交job6
        res5.get()
        res6 = pool.apply_async(job, args=("job6", 2))
        
        # 等待所有任务结果返回
        res1.get()
        res2.get()
        res6.get()
    
    total_time = time.time() - start_time
    print(f"所有任务完成，总耗时: {total_time:.2f}秒")

关键说明： #

• apply_async用于异步提交单个任务，返回AsyncResult对象，调用get()会阻塞直到任务完成，正好用来控制依赖顺序。
• 使用with语句管理进程池，自动完成池的关闭和剩余任务等待。
• 之前用map_async直接打印完成，是因为未调用get()等待结果，主线程直接执行后续代码，误以为任务完成。

问题根源总结 #

• Process方法：未对所有进程执行join()，主线程提前计算耗时并打印。
• Pool的map_async：map_async是异步批量提交，且未调用get()等待结果，导致主线程直接执行后续打印逻辑，实际任务还在后台运行。

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