如何将OpenCV+Python线程版摄像头持续采集改写为多进程版本

实现说明 #

针对多线程版本受Python GIL锁限制、高分辨率采集效率低、故障硬件易阻塞主程序的问题，以下实现完全兼容原有线程版的调用接口，采用独立采集进程+共享内存传帧的方案，进程间无序列化拷贝开销，主程序可随时读取最新帧，采集进程异常不会卡死主流程。
核心设计逻辑：

• 摄像头实例完全在采集子进程内初始化、持有，规避OpenCV视频句柄跨进程传递的兼容性问题
• 采用共享内存存储最新帧数据，相比multiprocessing.Queue减少全量帧拷贝开销，适配4K等高分辨率采集场景
• 跨进程事件信号控制采集启停，状态同步可靠
• 采集过程遇读帧故障自动重试，始终保留最后一帧有效数据，适配硬件不稳定场景

import cv2
import numpy as np
from multiprocessing import Process, Event
from multiprocessing import shared_memory

class WebcamStream:
    def __init__(self, stream_id=0, width=3840, height=2880):
        self.stream_id = stream_id
        self.frame_width = width
        self.frame_height = height
        # 单帧BGR格式占用内存大小：宽*高*3通道
        self.frame_size = self.frame_width * self.frame_height * 3
        # 跨进程控制信号
        self.stopped = Event()
        self.cam_ready = Event()
        self.cam_open_failed = Event()
        # 进程、共享内存占位
        self.proc = None
        self.shm = None
        self.shm_name = None
        self.frame = None

    def _collect_frame(self):
        # 子进程内初始化摄像头，避免句柄跨进程传递异常
        camera = cv2.VideoCapture(self.stream_id)
        camera.set(cv2.CAP_PROP_FRAME_WIDTH, self.frame_width)
        camera.set(cv2.CAP_PROP_FRAME_HEIGHT, self.frame_height)

        if not camera.isOpened():
            self.cam_open_failed.set()
            camera.release()
            return

        # 关联共享内存，写入首帧
        shm = shared_memory.SharedMemory(name=self.shm_name)
        frame_buffer = np.ndarray((self.frame_height, self.frame_width, 3), dtype=np.uint8, buffer=shm.buf)
        ret, first_frame = camera.read()
        if not ret:
            self.cam_open_failed.set()
            camera.release()
            shm.close()
            return
        np.copyto(frame_buffer, first_frame)
        self.cam_ready.set()

        # 持续采集循环
        while not self.stopped.is_set():
            ret, frame = camera.read()
            if not ret:
                # 读帧失败跳过，保留最后一帧有效数据
                continue
            # 新帧覆盖共享内存旧数据
            np.copyto(frame_buffer, frame)

        # 子进程资源清理
        camera.release()
        shm.close()

    def start(self):
        self.stopped.clear()
        # 创建共享内存存储帧数据
        self.shm = shared_memory.SharedMemory(create=True, size=self.frame_size)
        self.shm_name = self.shm.name
        # 本地映射共享内存为numpy数组，无拷贝
        self.frame = np.ndarray((self.frame_height, self.frame_width, 3), dtype=np.uint8, buffer=self.shm.buf)
        # 启动守护采集进程
        self.proc = Process(target=self._collect_frame, args=(), daemon=True)
        self.proc.start()
        # 等待摄像头初始化完成
        self.cam_ready.wait(timeout=10)
        if self.cam_open_failed.is_set():
            self.stop()
            print("[Exiting]: Error accessing webcam stream.")
            exit(0)
        return self

    def read(self):
        # 返回帧副本，避免主进程修改共享内存数据；无需修改帧可去掉copy()提升性能
        return self.frame.copy()

    def stop(self):
        self.stopped.set()
        if self.proc is not None and self.proc.is_alive():
            self.proc.join(timeout=2)
            self.proc.terminate()
        # 释放共享内存
        if self.shm is not None:
            self.shm.close()
            self.shm.unlink()
            self.shm = None

调用示例 #

接口与原有线程版完全一致，原有业务代码无需修改即可直接替换：

if __name__ == "__main__":
    main_camera_stream = WebcamStream(stream_id=0)
    main_camera_stream.start()
    while True:
        frame = main_camera_stream.read()
        # 此处添加帧处理、显示等主程序逻辑
        cv2.imshow("camera", frame)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
    main_camera_stream.stop()
    cv2.destroyAllWindows()

性能提示：若主程序不需要对读取到的帧做原地修改，可将read方法中的.copy()去掉，直接返回共享内存映射的数组对象，可进一步降低帧读取开销。

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