Python multiprocessing.Process循环执行元素随机丢失问题求助

问题描述 #

以下代码通过multiprocessing.Process实现函数执行超时终止，循环传入参数并收集输出，但运行时print输出不稳定，有时能得到全部3条输出，有时仅能得到2条：

import multiprocessing
import time

def foo(arg, return_dict):
    """
    The function does something and modify return_dict.
    It does not return anything. 
    """
    print("The input arg is: " + arg)
    
    # Do Something that maybe time consuming
    
    return_dict['a'] =  # Some value
    return_dict['b'] =  # Some other value

def process(arg):
    """
    The function passes parameter to multiprocessing.Process.
    The execution will stop if exceeds 10 seconds. 
    """
    manager = multiprocessing.Manager()
    return_dict = manager.dict()
    p = multiprocessing.Process(target=foo, args=(arg, return_dict))
    p.start()
    time.sleep(10)
    p.terminate()
    p.join()
    
    return return_dict.values()

def main(parameter_list):
    """
    This is the main function. It uses a for loop to pass parameter 
    to process(), which execute foo with the parameter within the time
    limit. 
    """
    for arg in parameter_list:
        _ = process(arg)
        
if __name__ == '__main__':
    parameter_list = ['A', 'B', 'C']
    main(parameter_list)

预期输出：

The input arg is: A
The input arg is: B
The input arg is: C

实际输出存在随机性，偶尔缺失某条记录，且无法稳定复现。运行环境为Windows，无法使用signal实现超时。

核心原因分析 #

1. Windows子进程stdout缓冲机制：Windows系统中，子进程的print输出默认带有缓冲，当子进程被terminate强制终止时，缓冲区内的输出内容可能还未刷新到父进程的控制台，导致部分输出丢失。
2. 固定sleep的终止时机问题：代码中用time.sleep(10)固定等待10秒后终止进程，无论子进程是否已经完成执行。若子进程刚完成print但缓冲未刷新，就被强制终止，输出内容会丢失；若sleep期间子进程已经完成，多余的terminate操作也可能干扰输出缓冲的刷新流程。
3. Manager创建的资源开销：每次调用process函数都新建一个multiprocessing.Manager，Manager本身会启动子进程，频繁创建销毁可能导致系统资源竞争，间接影响子进程的输出稳定性。

解决建议与优化代码 #

1. 强制刷新输出缓冲 #

修改foo函数中的print语句，添加flush=True参数，强制将输出立即写入控制台，避免缓冲丢失：

print("The input arg is: " + arg, flush=True)

2. 用join超时替代固定sleep #

使用p.join(timeout=10)替代time.sleep(10)，只有当进程在超时后仍未结束时才调用terminate，避免终止已完成的进程：

def process(arg, manager=None):
    return_dict = manager.dict() if manager else multiprocessing.Manager().dict()
    p = multiprocessing.Process(target=foo, args=(arg, return_dict))
    p.start()
    # 等待10秒，若进程未结束则终止
    if not p.join(10):
        p.terminate()
        p.join()  # 确保进程资源释放
    
    return return_dict.values()

3. 复用Manager减少资源开销 #

在main函数中创建一次Manager，传入process函数复用，避免频繁创建子进程：

def main(parameter_list):
    with multiprocessing.Manager() as manager:
        for arg in parameter_list:
            _ = process(arg, manager)

完整优化后代码 #

import multiprocessing
import time

def foo(arg, return_dict):
    print("The input arg is: " + arg, flush=True)
    
    # Do Something that maybe time consuming
    
    return_dict['a'] = 1  # 示例值
    return_dict['b'] = 2  # 示例值

def process(arg, manager):
    return_dict = manager.dict()
    p = multiprocessing.Process(target=foo, args=(arg, return_dict))
    p.start()
    if not p.join(10):
        p.terminate()
        p.join()
    
    return return_dict.values()

def main(parameter_list):
    with multiprocessing.Manager() as manager:
        for arg in parameter_list:
            _ = process(arg, manager)
        
if __name__ == '__main__':
    parameter_list = ['A', 'B', 'C']
    main(parameter_list)

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