Python multiprocessing Queue在Worker进程被杀后失效的原因

我用Python的multiprocessing.Pool和Queue写了一个模拟邮件管理的程序，初始运行正常，但手动杀死一个Worker进程后，Pool自动重建的新Worker卡在msg_ids.get()无法从队列获取数据，环境是Ubuntu 22.04 LTS、Python 3.10.4。

程序代码如下：

from multiprocessing import Pool, Queue
import time
import uuid
import os


NB_WORKERS = 3
NB_MAILS_PER_5_SECONDS = 2
MAIL_MANAGEMENT_DURATION_SECONDS = 1


def list_new_mails_id():
    for i in range(NB_MAILS_PER_5_SECONDS):
        # fake mailbox msg list
        yield str(uuid.uuid1())


def mail_worker(msg_ids):
    pid = os.getpid()
    print(f"Starting worker PID = {pid} and queue={msg_ids} queue size={msg_ids.qsize()} ...")
    while True:
        print(f"[{pid}] Waiting a mail to manage...")
        msg_id = msg_ids.get()
        print(f"[{pid}] managing mail msg_id = {msg_id} ...")
        # here should read mail msg_id and remove it from mailbox when finish
        print(f"[{pid}] --> fake duration of {MAIL_MANAGEMENT_DURATION_SECONDS}s")
        time.sleep(MAIL_MANAGEMENT_DURATION_SECONDS)


if __name__ == "__main__":
    msg_id_queue = Queue()

    with Pool(NB_WORKERS, mail_worker, (msg_id_queue,)) as p:
        while True:            
            for msg_id in list_new_mails_id():
                msg_id_queue.put(msg_id)
            print("\nWaiting for new mails to come...\n")
            time.sleep(5)

原因分析 #

问题核心在于multiprocessing.Queue的底层实现缺陷：

• Queue基于**管道（pipe）**和内置锁、信号量实现进程间同步，这些同步原语的状态完全依赖Worker进程正常维护。
• 当Worker在执行get()操作时被强制杀死，极大概率会导致Queue内部的同步原语异常：比如Worker已经获取了Queue的内部锁但未释放，或者信号量计数逻辑混乱。
• Pool重启Worker后虽然传入同一个Queue对象，但Queue的内部异常状态无法自动恢复，新Worker调用get()时会因等待锁、或误判队列无数据而永久阻塞。

解决方案 #

1. 改用multiprocessing.Manager().Queue()
   Manager维护的Queue由独立的管理进程负责同步逻辑，即使Worker异常退出，管理进程会自动修复Queue的同步状态，避免锁或计数异常。修改初始化代码：

   from multiprocessing import Pool, Manager
   
   if __name__ == "__main__":
       manager = Manager()
       msg_id_queue = manager.Queue()
   

2. 避免强制杀死Worker
   在Worker中监听SIGTERM等终止信号，收到信号后完成当前任务再优雅退出，从根源上避免Queue进入异常状态。
3. 改用Pool原生任务提交机制
   放弃Worker无限循环取队列的模式，改用Pool.apply_async()或Pool.map()提交任务，Pool会自动管理Worker生命周期和任务分配，Worker异常重启后不会出现同步问题。

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