如何通过管道将子进程函数生成的字典传递给父进程？

这问题我太熟了，之前帮不少开发者解决过类似的多进程数据传递痛点。核心要解决的就是把需要的字典数据和子进程的其他日志输出彻底隔离开，同时避免文件覆盖的问题。给你几个靠谱的实战方案，按需选：

方案1：分离标准输出（仅传结果）与标准错误（输出日志） #

这是改动最小、最容易维护的方案。思路很简单：把子进程的普通日志输出到stderr（标准错误流），只将序列化后的字典输出到stdout（标准输出流）。父进程只读取stdout的内容解析字典，stderr的日志可以选择忽略或单独保存，完全不会干扰结果解析。

改进后的 parent.py #

import subprocess
import json

# 启动子进程，同时捕获stdout和stderr
proc = subprocess.Popen(
    ['python3', 'child.py'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True  # 直接按文本处理，无需手动decode字节数据
)

# communicate()会等待子进程结束，返回(stdout内容, stderr内容)
stdout_data, stderr_data = proc.communicate()

# 解析stdout中的JSON字典
try:
    result_dict = json.loads(stdout_data)
    print("从子进程拿到的字典：", result_dict)
except json.JSONDecodeError as e:
    print(f"字典解析失败：{e}")
    print("子进程日志输出：", stderr_data)

改进后的 child.py #

import json
import sys

def child_function():
    result = {1: "one", 2: "two"}
    
    # 普通日志全部输出到stderr，不会污染stdout的结果
    print("子进程正在执行任务...", file=sys.stderr)
    print("任务执行完成，准备返回结果", file=sys.stderr)
    
    # 仅将序列化后的字典输出到stdout
    print(json.dumps(result))

if __name__ == "__main__":
    child_function()

这个方案完全不需要创建文件，Celery多实例运行时也不会有文件覆盖的问题，非常适合你的场景。

方案2：使用额外的专用管道（零侵入原有日志） #

如果你的子进程已经有大量print输出日志，不想逐个修改为sys.stderr，可以创建一个额外的管道，让子进程专门往这个管道写结果数据，彻底和日志输出隔离。

parent.py 代码 #

import subprocess
import json
import os

# 创建一对匿名管道：父进程持有读端，子进程持有写端
read_fd, write_fd = os.pipe()

# 启动子进程时，将写端的文件描述符作为参数传给子进程
proc = subprocess.Popen(
    ['python3', 'child.py', str(write_fd)],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    close_fds=False,  # 保留额外的文件描述符
    text=True
)

# 父进程关闭写端，避免子进程退出后管道未正常关闭
os.close(write_fd)

# 从专用管道读取子进程写入的结果
with os.fdopen(read_fd, 'r') as pipe_file:
    result_json = pipe_file.read()
    result_dict = json.loads(result_json)
    print("从专用管道拿到的字典：", result_dict)

# 可选：处理子进程的普通日志输出
stdout_log, stderr_log = proc.communicate()
print("子进程stdout日志：", stdout_log)
print("子进程stderr日志：", stderr_log)

child.py 代码 #

import json
import sys
import os

def child_function(result_fd):
    result = {1: "one", 2: "two"}
    
    # 原有print正常输出到stdout，完全不用修改
    print("这是子进程的普通日志，直接输出到stdout")
    
    # 将序列化后的字典写入专用管道
    with os.fdopen(result_fd, 'w') as pipe_file:
        pipe_file.write(json.dumps(result))

if __name__ == "__main__":
    # 从命令行参数获取父进程传递的管道写端文件描述符
    result_fd = int(sys.argv[1])
    child_function(result_fd)

这个方案对原有代码侵入性极低，每个子进程的管道都是独立的，多实例运行绝对不会冲突。

方案3：使用Pickle序列化（支持复杂Python类型） #

如果你的字典里包含JSON不支持的类型（比如datetime、自定义类实例），可以用Python的pickle模块序列化。注意：仅当子进程是完全可信的情况下使用这个方案，因为pickle反序列化存在安全风险，不可信的子进程可能通过恶意序列化数据执行任意代码。

parent.py 代码 #

import subprocess
import pickle

proc = subprocess.Popen(
    ['python3', 'child.py'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

# pickle是二进制序列化，所以不用text=True
stdout_data, stderr_data = proc.communicate()
result_dict = pickle.loads(stdout_data)
print("拿到的字典：", result_dict)

child.py 代码 #

import pickle
import sys

def child_function():
    # 包含JSON不支持的类型，比如元组、datetime（这里用元组示例）
    result = {1: "one", 2: "two", 3: (1, 2, 3)}
    
    # 普通日志输出到stderr
    print("处理包含复杂类型的字典", file=sys.stderr)
    
    # 二进制序列化后输出到stdout
    sys.stdout.buffer.write(pickle.dumps(result))

if __name__ == "__main__":
    child_function()

总结 #

• 优先选方案1：改动最小、安全可靠，完全适配你的Celery多实例场景；
• 若不想修改原有日志代码，选方案2；
• 若需传递复杂Python类型，且子进程可信，选方案3。

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