OpenCV多流系统多线程与多进程问题求助

环境背景 #

• Python 3.10.6
• OpenCV（基于FFmpeg后端）
• 目标：搭建多流视频播放系统，采用多线程/多进程处理视频帧

触发的错误 #

运行程序或关闭单个视频窗口时，出现三类错误：

• OSError: handle is closed
• Assertion fctx->async_lock failed at libavcodec/pthread_frame.c:173
• [h264 @ 0000016d982b5c00] Invalid NAL unit size (0 > 2210)

完整报错栈 #

Exception in thread Thread-2 (__show_video):
Traceback (most recent call last):
  File "C:\Python 3.10.6\lib\threading.py", line 1016, in _bootstrap_inner
    self.run()
  File "C:\Python 3.10.6\lib\threading.py", line 953, in run
    self._target(*self._args, **self._kwargs)
  File "D:\Desktop\Opencv_multidisplay\main.py", line 28, in __show_video
    frame = self.__queue.get()
  File "C:\Python 3.10.6\lib\multiprocessing\queues.py", line 103, in get
    res = self._recv_bytes()
  File "C:\Python 3.10.6\lib\multiprocessing\connection.py", line 217, in recv_bytes
    self._check_closed()
  File "C:\Python 3.10.6\lib\multiprocessing\connection.py", line 141, in _check_closed
    raise OSError("handle is closed")
OSError: handle is closed

错误根源拆解 #

1. OSError: handle is closed
   多进程队列（multiprocessing.Queue）的生产者进程已退出，但消费者线程仍在尝试从已关闭的队列中取数据。关闭视频时未同步终止生产者、通知消费者，导致消费者持续调用queue.get()触发句柄错误。

2. FFmpeg相关断言与NAL单元错误
   OpenCV依赖FFmpeg解码视频，多线程/多进程环境下若共享VideoCapture实例，会导致FFmpeg解码上下文的线程安全问题（如async_lock断言失败）；Invalid NAL unit size多是因为进程退出时传输了不完整帧，或共享解码资源导致帧数据损坏。

解决方案与优化实现 #

1. 规范生命周期管理 #

为每个视频流绑定独立的生产者进程与消费者线程，通过事件标记控制运行状态，关闭时先终止生产者再同步消费者退出，避免队列访问异常。

2. 保证解码资源独立性 #

禁止多线程/多进程共享同一个cv2.VideoCapture对象，每个进程/线程单独初始化解码实例，从根源避免FFmpeg上下文冲突。

3. 优化后代码示例 #

import cv2
import threading
import multiprocessing
from queue import Empty

class VideoStream:
    def __init__(self, source):
        self.source = source
        self.frame_queue = multiprocessing.Queue(maxsize=10)
        self.running_flag = multiprocessing.Event()
        self.running_flag.set()
        
        # 启动帧读取进程（生产者）
        self.capture_proc = multiprocessing.Process(target=self._read_frames)
        self.capture_proc.start()
        
        # 启动帧显示线程（消费者）
        self.display_thread = threading.Thread(target=self._show_frames)
        self.display_thread.daemon = True
        self.display_thread.start()

    def _read_frames(self):
        cap = cv2.VideoCapture(self.source)
        while self.running_flag.is_set() and cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            # 队列满时丢弃旧帧，避免进程阻塞
            if not self.frame_queue.full():
                try:
                    self.frame_queue.put(frame, timeout=0.1)
                except:
                    pass
        cap.release()
        # 发送流结束信号
        self.frame_queue.put(None)

    def _show_frames(self):
        window_name = f"Stream: {self.source}"
        while True:
            try:
                frame = self.frame_queue.get(timeout=0.5)
                if frame is None:
                    break
                cv2.imshow(window_name, frame)
                # 监听窗口关闭或退出指令
                if cv2.waitKey(1) & 0xFF == ord('q'):
                    self.stop()
                    break
            except Empty:
                continue
        cv2.destroyWindow(window_name)

    def stop(self):
        self.running_flag.clear()
        # 等待进程终止，超时则强制结束
        if self.capture_proc.is_alive():
            self.capture_proc.join(timeout=2)
        if self.capture_proc.is_alive():
            self.capture_proc.terminate()

if __name__ == "__main__":
    # 示例：同时开启摄像头与本地视频文件
    stream_list = [
        VideoStream(0),
        VideoStream("video1.mp4"),
        VideoStream("video2.mp4")
    ]
    try:
        for stream in stream_list:
            stream.display_thread.join()
    except KeyboardInterrupt:
        pass
    finally:
        for stream in stream_list:
            stream.stop()
        cv2.destroyAllWindows()

代码关键说明 #

• 每个视频流独立进程负责帧读取，线程负责显示，彻底隔离解码资源
• 用multiprocessing.Event控制生产者运行状态，关闭时优雅终止
• 队列满时丢弃旧帧，防止生产者阻塞；通过None信号通知消费者流结束
• 单独创建VideoCapture实例，解决FFmpeg解码上下文的线程安全问题

额外注意事项 #

• 若采用多线程替代多进程，需将multiprocessing.Queue替换为queue.Queue，且每个线程单独初始化VideoCapture
• 关闭窗口时必须先调用stop()终止生产者，再处理窗口销毁，避免残留进程或线程

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