OpenCV+Python实现持续读帧批量处理，解决多进程丢帧问题

问题分析 #

当前丢帧主要来自三个核心原因：

1. 读帧进程无差别丢帧：原代码在读取帧前就判断队列长度，超过100就弹出旧帧，读帧速度远快于处理速度时，大量未处理帧被提前丢弃。
2. 剩余帧未被处理：视频结束时，队列中不足100帧的批次会被直接忽略，无法进入处理流程。
3. 进程间通信开销：mp.Queue传递OpenCV帧（numpy数组）时需要序列化（pickle），额外消耗时间，加剧处理滞后。

具体解决方案 #

1. 修正读帧进程的帧丢弃逻辑

读取帧后再处理队列满的情况，仅当队列无法容纳新帧时，丢弃最早的旧帧，避免无意义丢帧：

def image_put(vidpath, q, end_event):
    cap = cv2.VideoCapture(vidpath)
    if not cap.isOpened():
        print("Read failed")
        return

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        
        # 队列满时丢弃最早的帧，保证新帧能入队
        if q.full():
            q.get()
        q.put(frame)

    cap.release()
    end_event.set()

2. 处理剩余不足批量的帧

在处理进程循环结束后，强制处理队列中剩余的未达100帧的批次，避免最后一批帧丢失：

from multiprocessing import Empty

def image_get(q, end_event, result_q):
    cnt = 0
    frame_batch = []

    while not end_event.is_set() or not q.empty():
        try:
            # 加超时避免死等空队列
            frame = q.get(timeout=0.1)
        except Empty:
            continue
        
        frame_batch.append(frame)

        if len(frame_batch) == 100:
            cnt += 100
            candidates_points = process_frames(frame_batch)
            result_q.put(candidates_points)
            frame_batch = []
    
    # 处理循环结束后剩余的帧
    if frame_batch:
        cnt += len(frame_batch)
        candidates_points = process_frames(frame_batch)
        result_q.put(candidates_points)
    
    print(f"{cnt} frames were processed")

3. 优化进程间通信效率

mp.Queue的序列化开销较大，改用共享内存传递帧，减少数据拷贝时间。示例如下：

import numpy as np
from multiprocessing import Array, Semaphore

def image_put(vidpath, shared_arr, shape, sem, end_event):
    cap = cv2.VideoCapture(vidpath)
    if not cap.isOpened():
        print("Read failed")
        return
    
    frame_np = np.frombuffer(shared_arr.get_obj(), dtype=np.uint8).reshape(shape)
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        
        # 等待处理进程读取完上一帧
        sem.acquire()
        # 将帧写入共享内存
        frame_np[:] = frame[:]
        # 通知处理进程有新帧
        sem.release()
        
    cap.release()
    end_event.set()

# 处理进程从共享内存读取帧（需配合信号量同步，批量逻辑需调整）
def image_get(shared_arr, shape, sem, end_event, result_q):
    cnt = 0
    frame_batch = []
    frame_np = np.frombuffer(shared_arr.get_obj(), dtype=np.uint8).reshape(shape)
    
    while not end_event.is_set():
        sem.acquire()
        # 复制共享内存中的帧到本地
        frame = frame_np.copy()
        sem.release()
        
        frame_batch.append(frame)
        if len(frame_batch) == 100:
            cnt += 100
            candidates_points = process_frames(frame_batch)
            result_q.put(candidates_points)
            frame_batch = []
    
    # 处理剩余帧
    if frame_batch:
        cnt += len(frame_batch)
        candidates_points = process_frames(frame_batch)
        result_q.put(candidates_points)

4. 增加处理进程数量

单进程处理速度不足时，启动多个处理进程并行处理帧批次：

def run_single_cam(vidpath):
    queue = mp.Queue(maxsize=1000)
    result_queue = mp.Queue()
    end_event = mp.Event()
    
    # 1个读帧进程 + 2个处理进程
    processes = [
        mp.Process(target=image_put, args=(vidpath, queue, end_event)),
        mp.Process(target=image_get, args=(queue, end_event, result_queue)),
        mp.Process(target=image_get, args=(queue, end_event, result_queue))
    ]

    [process.start() for process in processes]
    [process.join() for process in processes]

    # 后续处理结果队列
    while not result_queue.empty():
        # ... post processing

5. 调整批量处理策略

若必须每100帧处理一次，可改为边读边处理：读帧进程每积累100帧就直接传递给处理进程，而非等待队列攒满，减少帧积压。

额外建议 #

• 对于实时摄像头流，若处理速度仍跟不上，可考虑跳过部分帧（比如每2帧读1帧），但会牺牲精度，仅作为最后手段。
• 优先优化process_frames函数本身：比如使用OpenCV GPU加速（cv2.cuda）、减少冗余计算，这是从根源提升处理速度的核心。

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