Python 3.9多进程代码耗时远超2.7的原因及优化咨询

我有一段基于multiprocessing模块apply_async方法的代码，在Python 2.7中可正常运行，能并行处理任务并更新主GUI。升级到Python 3.9后，不仅原GUI更新功能失效，简化后的调试代码耗时从约48.965秒暴涨至372.522秒（Windows 10环境）。

简化代码如下：

import multiprocessing
import time
import datetime


def main():
    start_time = datetime.datetime.now()
    print('Spinning up pool')
    pool = multiprocessing.Pool(processes=10)
    vals = range(100)
    results = []
    print('Adding processes')
    runs = [pool.apply_async(calc, (x, 1), callback=results.append) for x in vals]

    print('Working...')
    while len(vals) != len(results):
        print('Results: {}'.format(results))
        time.sleep(1)

    pool.close()
    pool.join()
    print('Done')
    end_time = datetime.datetime.now()
    duration = end_time - start_time
    print('Program took {} seconds to complete'.format(duration.total_seconds()))

def calc(x, y):
    print(x + y)
    time.sleep(2)
    return(x+y)

if __name__ == "__main__":
    main()

耗时差异的核心原因 #

• Windows进程启动机制与管道阻塞：Python 3.4+在Windows上默认采用spawn方式创建子进程（Python 2.7采用旧的win32进程创建逻辑），子进程的stdout通过管道与主进程关联。代码中calc函数包含print语句，当主进程陷入while循环的time.sleep(1)时，无法及时读取管道中的子进程输出，导致管道缓冲区被填满，子进程卡在print调用处无法继续执行，最终任务从并行退化为串行，耗时大幅增加。
• 低效的结果轮询逻辑：原代码通过轮询len(results)判断任务是否完成，每次sleep(1)会阻塞主进程1秒，既浪费CPU时间，又加剧了管道阻塞问题，进一步拖慢整体效率。

代码优化方法 #

方法1：移除不必要的子进程输出 #

如果不需要子进程的print输出，直接删除calc中的print(x + y)语句，避免管道阻塞：

def calc(x, y):
    time.sleep(2)
    return x + y

方法2：优化结果等待逻辑，避免主进程长时间阻塞 #

替换低效的while轮询，改用AsyncResult的wait()方法等待所有任务完成：

def main():
    start_time = datetime.datetime.now()
    print('Spinning up pool')
    pool = multiprocessing.Pool(processes=10)
    vals = range(100)
    results = []
    print('Adding processes')
    runs = [pool.apply_async(calc, (x, 1), callback=results.append) for x in vals]

    print('Working...')
    # 等待所有异步任务完成
    for run in runs:
        run.wait()

    pool.close()
    pool.join()
    print('Done')
    end_time = datetime.datetime.now()
    duration = end_time - start_time
    print('Program took {} seconds to complete'.format(duration.total_seconds()))

方法3：使用更高效的批量任务接口 #

改用pool.map_async替代apply_async批量提交任务，代码更简洁且效率更高：

def main():
    start_time = datetime.datetime.now()
    print('Spinning up pool')
    pool = multiprocessing.Pool(processes=10)
    vals = range(100)
    print('Adding processes')
    # 批量提交任务，每个任务传入(x,1)
    result_obj = pool.map_async(lambda x: calc(x, 1), vals)

    print('Working...')
    # 获取所有结果（会自动等待任务完成）
    results = result_obj.get()

    pool.close()
    pool.join()
    print('Done')
    end_time = datetime.datetime.now()
    duration = end_time - start_time
    print('Program took {} seconds to complete'.format(duration.total_seconds()))

更优的任务等待与GUI更新方案 #

如果需要在任务执行过程中更新主GUI（如Tkinter、PyQt），推荐使用迭代式获取结果的方式，既避免阻塞主进程，又能实时处理完成的任务：

示例：使用pool.imap_unordered实时处理结果 #

def main():
    start_time = datetime.datetime.now()
    print('Spinning up pool')
    pool = multiprocessing.Pool(processes=10)
    vals = range(100)
    print('Working...')
    
    # 迭代获取完成的任务结果，无需提前等待全部完成
    for result in pool.imap_unordered(lambda x: calc(x, 1), vals):
        # 此处可添加GUI更新逻辑，如更新进度条、显示结果
        print(f'Received result: {result}')

    pool.close()
    pool.join()
    print('Done')
    end_time = datetime.datetime.now()
    duration = end_time - start_time
    print('Program took {} seconds to complete'.format(duration.total_seconds()))

这种方式下，主进程会在每次获取到一个完成的任务结果时立即处理（比如更新GUI），不会长时间阻塞，同时子进程的输出管道也能被及时处理，避免阻塞。

内容的提问来源于stack exchange，提问作者Das.Rot