为何Python信号处理器会阻塞queue？多线程下如何安全入队信号？

问题描述 #

我正在编写Python 3脚本，通过将信号（如signal.SIGWINCH）放入队列实现信号处理，同时有独立线程将用户输入放入同一队列，所有队列内容由主线程统一处理。但程序偶尔会突然挂起，排查发现问题出在信号处理器与queue.Queue的交互上：随机次数的信号入队后，信号处理器尝试向队列添加信号时会阻塞队列，后续任何与队列交互的线程（调用queue.put(block=False)、queue.get(block=False)或queue.empty()）都会随之挂起。即使使用非阻塞的队列操作，问题依然存在。

复现代码 #

在Python 3.13、Linux环境下，反复调整终端窗口触发SIGWINCH信号即可复现问题：

from queue import Queue, Empty
import signal

event_queue = Queue()

def signal_handler(signum, frame):
    event_queue.put(signum)

signal.signal(signal.SIGWINCH, signal_handler)

while True:
    try:
        print("Attempting to get an event from the queue...")
        evt = event_queue.get(block=False)
        print("Successfully got event from the queue.")
    except Empty:
        print("The queue is empty, try again.")

异常现象 #

程序会在调用queue.get(block=False)时无限挂起，尽管已经指定block=False，典型输出如下：

(...)
Attempting to get an event from the queue...
The queue is empty, try again.
Attempting to get an event from the queue...
(code hangs indefinitely)

问题原因 #

queue.Queue的线程安全基于内部的threading.Lock和threading.Condition实现，但信号处理器的执行时机是异步的，它可能在队列内部锁持有期间被触发。

当主线程正在执行队列操作（比如get(block=False)）并持有锁时，信号处理器被触发，此时信号处理器尝试调用event_queue.put()——这会尝试获取同一个锁。但信号处理器是在主线程的上下文里执行的，主线程此时正持有锁，又因为信号处理是中断式执行，主线程无法释放锁，最终导致死锁：信号处理器等待锁释放，而持有锁的主线程被信号中断，无法继续执行释放锁的逻辑，整个队列的所有操作都被阻塞。

即使使用block=False参数，队列操作依然需要先获取内部锁才能执行判断，所以同样会触发死锁。

安全的信号入队方案 #

方案1：用threading.Event中转信号 #

创建专门的信号处理线程，信号处理器仅做信号记录和事件通知，由独立线程完成队列操作，避免在信号上下文里持有队列锁：

from queue import Queue, Empty
import signal
import threading

event_queue = Queue()
signal_event = threading.Event()
pending_signals = []
signal_lock = threading.Lock()

def signal_handler(signum, frame):
    with signal_lock:
        pending_signals.append(signum)
    signal_event.set()

def signal_worker():
    while True:
        signal_event.wait()
        with signal_lock:
            signals = pending_signals.copy()
            pending_signals.clear()
        signal_event.clear()
        for sig in signals:
            event_queue.put(sig)

signal.signal(signal.SIGWINCH, signal_handler)
threading.Thread(target=signal_worker, daemon=True).start()

while True:
    try:
        print("Attempting to get an event from the queue...")
        evt = event_queue.get(block=False)
        print("Successfully got event from the queue.")
    except Empty:
        print("The queue is empty, try again.")

方案2：使用queue.SimpleQueue（Python 3.7+） #

queue.SimpleQueue是简化版线程安全队列，仅依赖threading.Lock、不使用Condition，内部逻辑更简洁，触发死锁的概率极低：

from queue import SimpleQueue, Empty
import signal

event_queue = SimpleQueue()

def signal_handler(signum, frame):
    event_queue.put(signum)

signal.signal(signal.SIGWINCH, signal_handler)

while True:
    try:
        print("Attempting to get an event from the queue...")
        evt = event_queue.get(block=False)
        print("Successfully got event from the queue.")
    except Empty:
        print("The queue is empty, try again.")

方案3：使用multiprocessing.Queue（谨慎选择） #

multiprocessing.Queue基于管道实现，操作不依赖线程锁，而是进程间通信机制，信号处理器中调用put()不会触发死锁。但该方案存在进程间通信的性能开销，适合对性能要求不高的场景。

关键结论 #

• 永远不要在信号处理器中直接操作queue.Queue（即使是非阻塞操作），信号的异步执行特性可能导致队列内部锁死锁。
• 安全的做法是让信号处理器只做最少量的工作（比如设置事件、记录信号），将队列操作交给常规线程完成。

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