Python multiprocessing中如何从队列子进程获取错误标识与消息

多进程拥有独立的内存空间，你在consumer子进程内定义的error_flag是子进程私有变量，主进程无法直接访问，必须通过进程间通信(IPC)机制传递错误信息。下面提供两种可行实现方案：

方案1：新增错误队列传递错误信息 #

这种方式灵活性最高，除了错误标识还可以传递出错的命令、退出码、错误描述等自定义信息，是最推荐的做法。
修改后的核心代码如下：

import os
import time
from multiprocessing import Process, Queue, Lock

command_queue = Queue()
error_queue = Queue() # 新增错误队列
lock = Lock()
consumers = []
consumer_num = 3 # 示例值，可根据你的需求修改
test_config_list_path = [] # 你的配置路径列表，按需填充

def producer(queue, lock, test_config_list_path):
    for config_path in test_config_list_path:
        # 这里的process_to_be_queued是你要执行的命令，自行替换
        queue.put((config_path, process_to_be_queued))
    # 任务推送完成后，放和consumer数量相同的None作为结束标识
    for _ in range(consumer_num):
        queue.put(None)

def consumer(queue, lock, error_queue):
    while True:
        elem = queue.get()
        if elem is None:
            return
        status = os.system(elem[1])
        exit_code = os.WEXITSTATUS(status) # 提取真实退出码
        if exit_code != 0:
            # 出错时把错误信息推送到错误队列
            error_queue.put({
                "config_path": elem[0],
                "command": elem[1],
                "exit_code": exit_code
            })

# 进程初始化
p = Process(target=producer, args=(command_queue, lock, test_config_list_path))
for i in range(consumer_num):
    c = Process(target=consumer, args=(command_queue, lock, error_queue))
    consumers.append(c)

p.daemon = True
p.start()

for c in consumers:
    c.daemon = True
    c.start()

p.join()
for c in consumers:
    c.join()

# 主进程读取错误队列判断是否有错误
error_list = []
while not error_queue.empty():
    error_list.append(error_queue.get())

if error_list:
    # 你的错误处理逻辑
    print(f"共发现{len(error_list)}个执行错误：{error_list}")
    Stop_this_process_and_send_a_message!

方案2：使用共享变量传递错误标识 #

如果你只需要知道是否有错误、不需要具体错误详情，可以用multiprocessing.Value创建全局共享的错误标识，注意读写时要加锁避免并发冲突：

import os
import time
from multiprocessing import Process, Queue, Lock, Value

command_queue = Queue()
lock = Lock()
error_flag = Value('i', 0) # 初始化整型共享变量，0表示无错误，1表示有错误
consumers = []
consumer_num = 3
test_config_list_path = []

def producer(queue, lock, test_config_list_path):
    for config_path in test_config_list_path:
        queue.put((config_path, process_to_be_queued))
    for _ in range(consumer_num):
        queue.put(None)

def consumer(queue, lock, error_flag):
    while True:
        elem = queue.get()
        if elem is None:
            return
        status = os.system(elem[1])
        exit_code = os.WEXITSTATUS(status)
        if exit_code != 0:
            # 写共享变量前加锁
            with lock:
                error_flag.value = 1

# 进程初始化逻辑和之前一致，只需要把error_flag传给consumer即可
p = Process(target=producer, args=(command_queue, lock, test_config_list_path))
for i in range(consumer_num):
    c = Process(target=consumer, args=(command_queue, lock, error_flag))
    consumers.append(c)

p.daemon = True
p.start()

for c in consumers:
    c.daemon = True
    c.start()

p.join()
for c in consumers:
    c.join()

# 主进程直接读取共享变量值
if error_flag.value == 1:
    Stop_this_process_and_send_a_message!

注意事项 #

• 你原代码的consumer没有退出逻辑，必须在producer所有任务推送完成后，往command_queue里放入和consumer数量相等的None作为结束信号，否则consumer会一直阻塞在queue.get()调用，无法退出
• os.system的返回值不是直接的命令退出码，Linux系统下需要调用os.WEXITSTATUS(status)提取真实的退出码，Windows下可以直接判断返回值是否为0

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