OpenCV+Python多进程相机持续采集：按需返回单帧实现（续问）

背景 #

基于OpenCV、Python多进程实现USB摄像头采集时，原有框架已经实现了子进程持续采集、共享内存传输帧的能力，需要改造为主进程主动发起请求时才返回单帧的按需取帧模式。

核心改造逻辑 #

• 新增控制通道：用一个专属队列传输主进程的取帧请求，用multiprocessing.Event做帧就绪的同步信号，仅传输极小体积的控制信号，额外开销可以忽略
• 保留后台持续采集逻辑：子进程始终循环读取摄像头画面写入共享内存，保证共享内存中始终存放最新的实时帧，不会出现帧积压、延迟高的问题
• 同步读写控制：子进程收到取帧请求、写完当前最新帧后，通过就绪事件通知主进程读取，避免主进程读到半写入的损坏帧
• 资源自动回收：进程退出时自动释放摄像头句柄、共享内存段，避免设备占用、内存泄漏问题

改造后完整代码 #

import numpy as np
import cv2
from multiprocessing import Process, Queue, Event
from multiprocessing.shared_memory import SharedMemory

def produce_frames(info_q, request_q, frame_ready_event):
    # 初始化摄像头
    cap = cv2.VideoCapture(0)
    success, frame = cap.read()
    if not success:
        raise RuntimeError("摄像头初始化失败，无法读取首帧")
    
    # 初始化共享内存
    shm = SharedMemory(create=True, size=frame.nbytes)
    framebuffer = np.ndarray(frame.shape, frame.dtype, buffer=shm.buf)
    framebuffer[:] = frame
    
    # 把共享内存信息、帧参数传回主进程
    info_q.put(shm.name)
    info_q.put(frame.shape)
    info_q.put(frame.dtype)

    try:
        while True:
            # 持续读取最新帧写入共享内存
            ret, _ = cap.read(framebuffer)
            if not ret:
                break
            # 检查是否有取帧请求
            if not request_q.empty():
                try:
                    request_q.get_nowait()
                    # 通知主进程帧已就绪
                    frame_ready_event.set()
                except:
                    pass
    except KeyboardInterrupt:
        pass
    finally:
        cap.release()
        shm.close()
        shm.unlink()

def get_latest_frame(request_q, frame_ready_event, framebuffer):
    # 重置就绪信号
    frame_ready_event.clear()
    # 发送取帧请求
    request_q.put(True)
    # 等待帧写入完成
    frame_ready_event.wait()
    # 返回帧副本，避免后续新帧覆盖
    return framebuffer.copy()

if __name__ == "__main__":
    # 信息传输队列、取帧请求队列、帧就绪事件
    info_q = Queue()
    request_q = Queue()
    frame_ready_event = Event()

    # 启动采集子进程
    producer = Process(target=produce_frames, args=(info_q, request_q, frame_ready_event))
    producer.start()

    # 主进程初始化共享内存映射
    shm_name = info_q.get()
    shape = info_q.get()
    dtype = info_q.get()
    shm = SharedMemory(name=shm_name)
    framebuffer = np.ndarray(shape, dtype, buffer=shm.buf)

    try:
        while True:
            # 示例：按任意键请求一帧，按q退出
            key = cv2.waitKey(0)
            if key == ord('q'):
                break
            # 按需取帧
            frame = get_latest_frame(request_q, frame_ready_event, framebuffer)
            cv2.imshow("按需取帧窗口", frame)
    except KeyboardInterrupt:
        pass
    finally:
        # 资源回收
        producer.terminate()
        producer.join()
        shm.close()
        cv2.destroyAllWindows()

使用注意事项 #

• 取帧时默认返回共享内存的副本，如果你不需要保留帧做后续耗时处理（比如推理、存盘），可以去掉.copy()直接读取framebuffer，性能更高，但要注意读完立刻使用，不要留存对象引用
• 采集子进程没有加任何等待逻辑，会始终以摄像头支持的最高帧率刷新共享内存，每次请求拿到的都是当前最新帧，不会出现旧帧排队的问题
• 可以根据业务逻辑自由调整主进程的取帧频率，不管是1秒取1帧还是10秒取1帧，子进程都会在后台正常刷新，不会阻塞摄像头运行

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