如何通过Python多线程分离YOLO ANPR检测与视频播放以提速

完全可以通过Python多线程分离视频播放与ANPR算法处理，以此缓解卡顿。核心逻辑是将帧读取、ANPR计算、视频渲染拆分为独立任务，用队列做线程间的安全数据传递，避免计算密集的ANPR阻塞视频播放线程。

实现步骤 #

• 用queue.Queue构建双缓冲区：一个存原始帧，一个存ANPR处理后的带标注帧，平衡各环节的速度差
• 帧读取线程：持续从视频/摄像头拉取帧，存入原始帧队列，不占用主线程资源
• ANPR处理线程：从原始帧队列取帧，运行YOLO检测+OCR识别，完成标注后存入结果队列
• 主线程：专注从结果队列取帧并渲染播放，保证播放流畅

代码示例 #

import cv2
import threading
from queue import Queue
from ultralytics import YOLO
import easyocr

# 初始化队列，限制容量防止内存溢出
raw_frame_queue = Queue(maxsize=8)
processed_frame_queue = Queue(maxsize=8)

# 加载预训练模型
plate_detector = YOLO("plate_detector.pt")
ocr_reader = easyocr.Reader(["zh-cn", "en"])  # 根据车牌语言调整

def frame_reader(video_source):
    cap = cv2.VideoCapture(video_source)
    # 可选：设置摄像头缓冲区大小，减少帧丢失
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 2)
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        if not raw_frame_queue.full():
            raw_frame_queue.put(frame)
    cap.release()
    raw_frame_queue.put(None)  # 标记读取结束

def anpr_processor():
    while True:
        frame = raw_frame_queue.get()
        if frame is None:
            break
        # YOLO检测车牌区域
        detection_results = plate_detector(frame, verbose=False)
        for res in detection_results:
            for box in res.boxes:
                x1, y1, x2, y2 = map(int, box.xyxy[0])
                # 裁剪车牌区域并做OCR
                plate_img = frame[y1:y2, x1:x2]
                ocr_res = ocr_reader.readtext(plate_img)
                if ocr_res:
                    plate_text = ocr_res[0][1]
                    # 在帧上绘制标注
                    cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
                    cv2.putText(frame, plate_text, (x1, y1-10), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0,255,0), 2)
        if not processed_frame_queue.full():
            processed_frame_queue.put(frame)
    processed_frame_queue.put(None)  # 标记处理结束

def video_player():
    while True:
        frame = processed_frame_queue.get()
        if frame is None:
            break
        cv2.imshow("Real-time ANPR", frame)
        # 按Q键退出
        if cv2.waitKey(1) & 0xFF == ord("q"):
            break
    cv2.destroyAllWindows()

if __name__ == "__main__":
    # 替换为视频路径或摄像头索引（比如0表示默认摄像头）
    source = "test_video.mp4"
    # 启动线程
    reader_thread = threading.Thread(target=frame_reader, args=(source,))
    processor_thread = threading.Thread(target=anpr_processor)
    
    reader_thread.start()
    processor_thread.start()
    
    # 主线程负责播放
    video_player()
    
    # 等待子线程结束
    reader_thread.join()
    processor_thread.join()

额外优化建议 #

• 队列大小根据硬件配置调整：内存充足可适当调大，避免帧丢失；内存有限则调小，防止内存占用过高
• 对输入帧降采样：比如用cv2.resize(frame, (640, 480))缩小帧尺寸，大幅降低ANPR计算量
• 若多线程仍性能不足，可改用multiprocessing多进程：绕过Python GIL限制，充分利用多核CPU处理ANPR计算
• 避免在处理线程中做任何UI操作，所有渲染逻辑统一放在主线程

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