使用Python multiprocessing模块时函数返回非预期结果问题

问题根因分析 #

• 第一类异常：返回None
  核心原因是some_systemcall函数存在死锁风险：使用subprocess.Popen绑定stdout=subprocess.PIPE时，先调用p.wait()再调用p.communicate()是错误写法。如果子进程输出内容超过管道默认缓冲区（通常64KB），子进程会阻塞等待主进程读取管道内容，主进程则阻塞在wait()等待子进程退出，形成死锁。此时子进程如果被系统OOM killer、资源限制规则或超时机制强制终止，worker进程会直接异常退出，任务没有正常返回值，回调函数拿到的result就是None。

• 第二类（空列表[]）、第三类异常（仅含result的列表）
  你提供的crunch_line是精简版本，省略的# stuff业务代码部分大概率存在不符合返回格式的提前返回分支：比如部分判断逻辑下直接返回空列表，或者直接返回some_systemcall的执行结果，没有带上i和file_name字段，你可以排查这部分省略代码的所有return路径。

修复方案 #

1. 修正some_systemcall的写法，删除p.wait()，直接调用p.communicate()（该方法会自动等待子进程结束，同时读取所有管道输出，避免死锁）：

def some_systemcall(line):
    call_list = ["program", line]
    p = subprocess.Popen(call_list, stdout=subprocess.PIPE,
                         stderr=subprocess.STDOUT)
    output = p.communicate()[0].decode('utf-8')
    if p.returncode != 0:
        raise SystemError(f"program Error:\n{output}")
    return output if output.strip() else "no result"

2. 给apply_async添加异常回调，捕获worker进程的未处理异常，避免异常返回值污染RESULTS：

# 新增异常回调函数
def log_error(error):
    RESULTS.append([None, None, f"子进程异常：{str(error)}"])

# 调用apply_async时添加参数
syndicate.append(pool.apply_async(crunch_line,
                                  [i, file_name, line],
                                  callback=log_result,
                                  error_callback=log_error))

3. 排查crunch_line中你省略的业务代码，确保所有return路径都返回[i, file_name, 结果]格式的列表。

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