Python多进程队列中worker1故障时如何阻塞生产者？

要实现当worker1故障时阻塞生产者的需求，核心是让生产者能感知到指定消费者的健康状态，根据状态动态控制生产节奏。以下是两种可行的实现方式：

方法1：进程状态检查+事件控制 #

通过监控worker1的存活状态和运行标记，让生产者在worker1故障时暂停生产：

1. 给worker1绑定一个运行状态事件，启动时标记为正常运行；
2. 生产者定期检查worker1的进程状态，若判定故障则阻塞等待。

示例代码：

import multiprocessing
import time

def worker1(queue, running_flag):
    try:
        running_flag.set()
        while True:
            item = queue.get()
            # 模拟故障：长时间休眠
            time.sleep(30)
            print(f"Worker1 processed: {item}")
            queue.task_done()
    except Exception:
        running_flag.clear()

def worker2(queue):
    while True:
        item = queue.get()
        print(f"Worker2 processed: {item}")
        queue.task_done()

def producer(queue, worker1_proc, running_flag):
    item_count = 0
    while True:
        # 检查worker1状态：存活但超过10秒无有效处理则判定故障
        if worker1_proc.is_alive():
            queue.put(item_count)
            print(f"Produced: {item_count}")
            item_count += 1
            time.sleep(1)
        else:
            print("Worker1故障，生产者阻塞中...")
            running_flag.wait()  # 等待worker1恢复

if __name__ == "__main__":
    queue = multiprocessing.JoinableQueue()
    running_flag = multiprocessing.Event()
    
    worker1_proc = multiprocessing.Process(target=worker1, args=(queue, running_flag))
    worker2_proc = multiprocessing.Process(target=worker2, args=(queue,))
    
    worker1_proc.start()
    worker2_proc.start()
    
    producer(queue, worker1_proc, running_flag)
    
    queue.join()
    worker1_proc.join()
    worker2_proc.join()

方法2：心跳检测机制 #

给worker1添加心跳发送逻辑，生产者通过心跳判断worker1是否正常，超时则阻塞：

1. 新增心跳队列，worker1定期发送心跳时间戳；
2. 生产者检查心跳间隔，超过阈值则暂停生产。

示例核心代码片段：

def worker1(queue, heartbeat_queue):
    while True:
        # 发送心跳
        heartbeat_queue.put(time.time())
        item = queue.get()
        # 模拟故障
        time.sleep(30)
        queue.task_done()

def producer(queue, heartbeat_queue):
    item_count = 0
    last_heartbeat = time.time()
    while True:
        try:
            # 非阻塞获取最新心跳
            last_heartbeat = heartbeat_queue.get(block=False)
        except multiprocessing.queues.Empty:
            pass
        
        # 10秒未收到心跳则阻塞
        if time.time() - last_heartbeat > 10:
            print("未收到Worker1心跳，生产者阻塞...")
            last_heartbeat = heartbeat_queue.get()
        
        queue.put(item_count)
        print(f"Produced: {item_count}")
        item_count += 1
        time.sleep(1)

关键注意点 #

• 故障判定逻辑需适配实际场景：比如区分进程死亡、死循环、长时间休眠等情况；
• 阻塞逻辑建议添加超时机制，避免永久死锁，方便手动干预恢复；
• 若worker1是死循环而非休眠，is_alive()会返回True，此时必须结合心跳或任务处理时间戳来判断状态。

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