Python多进程读取RTSP摄像头无缓存取最新帧的最优实现方案

原方案的核心问题是queue.empty()不是原子操作，判断与后续读写操作之间存在时间差，才会频繁触发竞态条件需要大量异常捕获。以下提供两种可直接落地的优化方案：

方案1：定长队列轻量优化（改动最小） #

仅需调整队列初始化参数和少量读写逻辑，完全靠队列原生原子操作避免无效判断，适配现有代码成本极低：

1. 初始化队列时设置容量为1：mp.Queue(maxsize=1)，天然限制最多只存1帧
2. 写端直接尝试非阻塞写入新帧，触发Full异常时先弹出旧帧再写入即可
3. 读端改为带超时的阻塞读取，避免空转占用CPU，同时可以及时响应停止信号

修改后核心代码如下：

launcher.py 调整 #

# 仅需修改队列初始化行，其余代码不变
frames_queue = mp.Queue(maxsize=1)

reader.py 调整 #

import cv2
import queue

def Reader(frames_queue):
    cap = cv2.VideoCapture('rtsp_address')
    while True:
        ret, frame = cap.read()
        if not ret:
            continue
        resized_frame = cv2.resize(frame, (1080, 720))
        try:
            frames_queue.put(resized_frame, block=False)
        except queue.Full:
            # 队列已满说明存在未处理旧帧，丢弃旧帧后写入新帧
            try:
                frames_queue.get_nowait()
                frames_queue.put(resized_frame, block=False)
            except (queue.Empty, queue.Full):
                # 极端竞态下直接跳过当前帧，不影响实时性
                pass

consumer.py 调整 #

import cv2
import queue

def Consumer(frames_queue, stop_switch):
    while not stop_switch.is_set():
        try:
            frame = frames_queue.get(timeout=0.1)
        except queue.Empty:
            continue
        # 执行帧处理逻辑
        cv2.imshow('output', cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        
        key = cv2.waitKey(1)
        if key == ord('q'):
            stop_switch.set()
            break

方案2：互斥锁+单值共享存储（无竞态，生产级稳定） #

由于需求仅需要保存最新一帧，队列属于冗余结构，直接用共享存储加互斥锁的方案可以彻底消除竞态问题，逻辑更清晰：

1. 用multiprocessing.Manager创建可跨进程共享的字典存储最新帧
2. 读写帧时都加互斥锁，保证同一时间只有一个进程操作数据
3. 不需要任何队列相关的异常捕获，代码复杂度极低

修改后核心代码如下：

launcher.py #

from reader import Reader
from consumer import Consumer
import multiprocessing as mp
from multiprocessing import set_start_method, Event, Lock
import sys

def main():
    set_start_method("spawn")
    with mp.Manager() as manager:
        # 共享存储，仅存最新1帧
        latest_frame = manager.dict()
        # 读写互斥锁
        frame_lock = Lock()
        stop_switch = Event()

        reader = mp.Process(target=Reader, args=(latest_frame, frame_lock), daemon=True)
        consumer = mp.Process(target=Consumer, args=(latest_frame, frame_lock, stop_switch), daemon=True)

        reader.start()
        consumer.start()

        while True:
            if stop_switch.is_set():
                reader.terminate()
                consumer.terminate()
                sys.exit(0)

if __name__ == "__main__":
    main()

reader.py #

import cv2

def Reader(latest_frame, frame_lock):
    cap = cv2.VideoCapture('rtsp_address')
    while True:
        ret, frame = cap.read()
        if not ret:
            continue
        resized_frame = cv2.resize(frame, (1080, 720))
        # 加锁直接替换最新帧
        with frame_lock:
            latest_frame['data'] = resized_frame

consumer.py #

import cv2

def Consumer(latest_frame, frame_lock, stop_switch):
    while not stop_switch.is_set():
        frame = None
        # 加锁读取最新帧，读完直接清空存储
        with frame_lock:
            if 'data' in latest_frame:
                frame = latest_frame.pop('data')
        
        if frame is not None:
            cv2.imshow('output', cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        
        key = cv2.waitKey(1)
        if key == ord('q'):
            stop_switch.set()
            break

如果需要处理高分辨率大帧降低跨进程拷贝开销，还可以将帧编码为JPEG格式后再存入共享存储，或者改用multiprocessing.SharedMemory直接操作内存块，性能会更高。

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