Python中使用Queue时子进程无法终止的问题求助

我刚接触多进程编程，现在想实现一个简单的相机模拟器——用子进程生成图像放到队列里，另一个子进程处理。但现在遇到个问题：初始化的相机模拟器子进程在调用camera.join()时会挂起/冻结，没法正常结束程序。

我的代码如下：

import multiprocessing
import queue
import time
import numpy as np

class CameraSimulator(multiprocessing.Process):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue
        self.connected = multiprocessing.Event()
        self.acquiring = multiprocessing.Event()

    def run(self):
        while self.connected.is_set():
            while self.acquiring.is_set():
                image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
                try:
                    self.queue.put(image)
                except queue.Full:
                    continue
                time.sleep(0.1)

    def connect(self):
        self.connected.set()
        self.start()

    def disconnect(self):
        self.connected.clear()

    def acquire(self):
        self.acquiring.set()

    def stop(self):
        self.acquiring.clear()

if __name__ == "__main__":

    image_queue = multiprocessing.Queue(maxsize=1000)
    camera = CameraSimulator(image_queue)

    print("Connect camera")
    camera.connect()

    print("Start camera acquisition")
    camera.acquire()
    time.sleep(2)
    print("Stop camera acquisition")
    camera.stop()

    time.sleep(2)

    print("Start camera acquisition again")
    camera.acquire()
    time.sleep(2)
    print("Stop camera acquisition")
    camera.stop()

    print("Disconnect camera")
    camera.disconnect()

    print("Draining queue")
    while not image_queue.empty():
        try:
            image_queue.get_nowait()
        except queue.Empty:
            break
    print("Queue drained")

    camera.join()
    print("Camera process terminated")

我怀疑是队列没清空的问题，所以特意加了排空队列的代码，但问题还是存在。会不会是因为多进程/多线程的特性，empty()方法判断队列是否为空并不靠谱？我可以用camera.terminate()强制终止子进程，但感觉这不是好做法，求各位大佬帮忙看看！

问题分析与解决思路 #

兄弟，我看了你的代码，核心问题不是队列排空，而是子进程被Queue.put()的阻塞调用卡住，没法响应退出信号，咱们一步步拆解：

1. 为什么子进程挂起？

你的run方法里，外层循环依赖connected事件判断是否退出，但如果子进程正卡在self.queue.put(image)的阻塞调用中，它根本没机会去检查connected的状态——put默认会一直阻塞直到队列有空闲位置，这就导致子进程永远退不出循环，join()自然会一直等待。

另外你提到的image_queue.empty()确实不靠谱，多进程环境下队列状态是动态的，你刚检查完“为空”，子进程可能又塞了数据进去，但这不是当前的核心矛盾。

2. 修复方案

给你两个关键修改点，保证子进程能正常退出：

（1）给Queue.put()加超时，避免永久阻塞

让put操作每隔一段时间就跳出阻塞，这样子进程就能回到循环里检查退出条件：

try:
    # 加0.05秒超时，到点就返回，不会一直卡着
    self.queue.put(image, timeout=0.05)
except queue.Full:
    continue

（2）调整循环逻辑，确保每次循环都能检查退出信号

把内层的while self.acquiring.is_set()改成if，或者让内层循环也能响应connected的变化，避免嵌套循环导致外层条件无法被及时检查：

def run(self):
    while self.connected.is_set():
        if self.acquiring.is_set():
            image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
            try:
                self.queue.put(image, timeout=0.05)
            except queue.Full:
                continue
            time.sleep(0.1)
        else:
            # 非采集状态下短暂休眠，减少空循环占用CPU
            time.sleep(0.1)

（3）额外优化：断开连接前先停止采集

虽然主进程里已经调用了stop()，但保险起见，可以在disconnect方法里先清除acquiring事件，避免子进程还在尝试生成图像：

def disconnect(self):
    self.acquiring.clear()  # 先停止采集
    self.connected.clear()

3. 完整修改后的代码

import multiprocessing
import queue
import time
import numpy as np

class CameraSimulator(multiprocessing.Process):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue
        self.connected = multiprocessing.Event()
        self.acquiring = multiprocessing.Event()

    def run(self):
        while self.connected.is_set():
            if self.acquiring.is_set():
                image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
                try:
                    # 加入超时，避免一直阻塞在put上
                    self.queue.put(image, timeout=0.05)
                except queue.Full:
                    continue
                time.sleep(0.1)
            else:
                # 非采集状态下短暂休眠，减少CPU占用
                time.sleep(0.1)

    def connect(self):
        self.connected.set()
        self.start()

    def disconnect(self):
        # 先停止采集，再断开连接
        self.acquiring.clear()
        self.connected.clear()

    def acquire(self):
        self.acquiring.set()

    def stop(self):
        self.acquiring.clear()

if __name__ == "__main__":

    image_queue = multiprocessing.Queue(maxsize=1000)
    camera = CameraSimulator(image_queue)

    print("Connect camera")
    camera.connect()

    print("Start camera acquisition")
    camera.acquire()
    time.sleep(2)
    print("Stop camera acquisition")
    camera.stop()

    time.sleep(2)

    print("Start camera acquisition again")
    camera.acquire()
    time.sleep(2)
    print("Stop camera acquisition")
    camera.stop()

    print("Disconnect camera")
    camera.disconnect()

    print("Draining queue")
    while True:
        try:
            # 用get_nowait循环取，直到队列为空，比empty()更可靠
            image_queue.get_nowait()
        except queue.Empty:
            break
    print("Queue drained")

    camera.join()
    print("Camera process terminated")

修改后再运行，子进程就能正常响应退出信号，join()也能顺利结束了。

备注：内容来源于stack exchange，提问作者thepero