Python多进程应用中capture_read_frame_process进程无法退出求助

按下显示窗口的'q'键后，窗口关闭，但程序卡在capture_read_frame_process.join()语句处，读取帧的进程无法终止。

相关代码 #

import cv2 as cv
import multiprocessing

class Camera(object):
    def __init__(self, source):
        self._source = source
        self.image_queue = multiprocessing.Queue()

        match self._source:
            case 'cam':
                self._source_url = 0
                self._apiPref = cv.CAP_DSHOW
            case 'boysr':
                self._source_url = 'rtsps://192.168.1.1:7441/xhw7D6R7BR8NP5vg'
                self._apiPref = cv.CAP_FFMPEG
            case 'frontd':   
                self._source_url = 'rtsps://192.168.1.1:7441/EOEohGh0eoXIWf28'
                self._apiPref = cv.CAP_FFMPEG
            case 'video':
                self._source_url = 'event.mp4'
                self._apiPref = cv.CAP_FFMPEG
    
    def read_frames(self, queue, stop_read):
        self.query_source = cv.VideoCapture(self._source_url, self._apiPref)
        self.query_source.set(cv.CAP_PROP_BUFFERSIZE, 1)
        self.query_source.set(cv.CAP_PROP_FPS, 20)

        if self.query_source.isOpened():
            print('Capture object successfully started')
        else:
            raise Exception('Unable to start capture object')

        _failFrame = 0
        while not stop_read.is_set():
            _hasFrame, _query_image = self.query_source.read()
            if not _hasFrame:
                _failFrame += 1
                if _failFrame == 15:
                    raise Exception(f"Failed to grab a frame after {_failFrame} consequtive tries")
                continue
            else:
                _failFrame = 0
                queue.put(_query_image)

        self.query_source.release()

    def show_frame(self, queue, stop_read):
        while True:
            _query_image = queue.get()
            cv.imshow('autoface', _query_image)
            if cv.waitKey(1) == ord('q'):
                cv.destroyAllWindows()
                break

if __name__ == '__main__':
    print('Starting capture object')
    capture = Camera('cam')

    # Create the event to use to stop the program
    stop_read = multiprocessing.Event()

    print('Starting read_frames process')
    capture_read_frame_process = multiprocessing.Process(target=capture.read_frames, args=(capture.image_queue, stop_read,))
    capture_read_frame_process.start()
    
    print('Starting show_frames process')
    capture_show_frame_process = multiprocessing.Process(target=capture.show_frame, args=(capture.image_queue, stop_read,))
    capture_show_frame_process.start()

    capture_show_frame_process.join()   # Should complete when 'q' is typed when the cv.imshow window is open and active.

    stop_read.set()     # This should then break the loop in the read_frames function of the Camera object
    
    capture_read_frame_process.join()
    
    print('\nProgram exited')

• 事件触发延迟：主进程在等待显示进程结束后才设置stop_read事件，此时读取进程可能卡在cv.VideoCapture.read()的阻塞调用中，无法及时响应停止信号。
• 队列无边界风险：未限制队列容量，读取进程可能持续写入帧导致内存占用过高；若后续调整队列大小，queue.put()可能因队列满而阻塞读取进程。
• 资源释放不严谨：读取进程中VideoCapture的释放逻辑未做异常处理，若进程抛出异常，可能导致设备资源无法正常释放。

1. 提前触发停止事件 #

修改show_frame函数，按下'q'后立即设置停止事件，让读取进程及时收到终止信号：

def show_frame(self, queue, stop_read):
    while True:
        # 处理队列空的情况，避免阻塞
        if not queue.empty():
            _query_image = queue.get()
            cv.imshow('autoface', _query_image)
        # 非阻塞监听按键
        key = cv.waitKey(1)
        if key == ord('q'):
            cv.destroyAllWindows()
            stop_read.set()  # 立即触发停止事件
            break

2. 优化读取进程的事件响应 #

在读取循环中优先检查停止信号，同时将资源释放放入finally块，确保异常场景下也能释放设备：

def read_frames(self, queue, stop_read):
    self.query_source = None
    try:
        self.query_source = cv.VideoCapture(self._source_url, self._apiPref)
        self.query_source.set(cv.CAP_PROP_BUFFERSIZE, 1)
        self.query_source.set(cv.CAP_PROP_FPS, 20)

        if not self.query_source.isOpened():
            raise Exception('Unable to start capture object')
        print('Capture object successfully started')

        _failFrame = 0
        while not stop_read.is_set():
            # 先检查停止信号，避免卡在read调用中
            if stop_read.is_set():
                break
            _hasFrame, _query_image = self.query_source.read()
            if not _hasFrame:
                _failFrame += 1
                if _failFrame == 15:
                    raise Exception(f"Failed to grab a frame after {_failFrame} consecutive tries")
                continue
            _failFrame = 0
            # 设置put超时，避免队列满时阻塞
            queue.put(_query_image, block=True, timeout=1)
    finally:
        if self.query_source is not None:
            self.query_source.release()

3. 限制队列容量 #

初始化队列时设置最大容量，避免内存溢出：

def __init__(self, source):
    self._source = source
    self.image_queue = multiprocessing.Queue(maxsize=2)  # 限制队列最多存2帧
    # ... 其余代码不变

4. 给主进程join设置超时 #

避免读取进程因异常无法终止时，主进程无限等待：

if __name__ == '__main__':
    # ... 其余代码不变
    capture_show_frame_process.join()
    stop_read.set()
    # 设置2秒超时，超时则强制终止进程
    capture_read_frame_process.join(timeout=2)
    if capture_read_frame_process.is_alive():
        capture_read_frame_process.terminate()
        capture_read_frame_process.join()
    print('\nProgram exited')

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