多摄像头Coco SSD目标检测多线程异常问题排查与解决咨询

问题描述 #

使用Coco SSD初始化目标检测模型，尝试对两个摄像头画面做检测，代码逻辑如下（注：代码存在基础错误）：

model=cv2.dnn_DetectionModel(weightsPath,configPath)
cam1=cv2.VideoCapture(0)
cam1=cv2.VideoCapture(1)  # 变量覆盖：cam1被重复赋值，cam2未定义

# thread 1
while True:
    success1,img1 = cam1.read()
    result1=model.detect(img1)
    cv2.imshow("result1", result1)  # 错误：detect返回值不是可直接显示的图像
    if cv2.waitKey(2) & 0xFF==ord('x'):
        break

# thread 2 
while True:
    success2,img2 = cam2.read()  # cam2未定义，会报错
    result2=model.detect(img2)
    cv2.imshow("result2", result2)
    if cv2.waitKey(2) & 0xFF==ord('x'):
        break

单线程运行时检测正常，但多线程运行时所有视频源均出现检测异常，无法正确识别目标。

问题原因 #

1. 模型实例非线程安全：cv2.dnn_DetectionModel的内部状态（如预处理缓冲区、推理计算资源）未做线程同步处理，多个线程同时调用model.detect()会引发数据竞争，导致推理结果混乱。
2. 代码基础错误：原代码中cam1被连续赋值两次，实际只初始化了摄像头1，摄像头0的实例被覆盖；同时cam2未定义，线程2会直接读取失败，干扰整体流程。
3. UI操作线程冲突：OpenCV的cv2.imshow()和cv2.waitKey()属于HighGUI模块操作，该模块并非完全线程安全，多线程下调用可能导致窗口渲染异常，甚至阻塞推理过程。

可行解决方案 #

方案1：为每个线程创建独立模型实例 #

每个线程单独初始化模型，避免共享内部状态，这是最稳定的方案：

import cv2
import threading

def detect_camera(cam_idx, window_name):
    # 每个线程独立初始化模型
    model = cv2.dnn_DetectionModel(weightsPath, configPath)
    # 配置模型参数（根据实际需求调整）
    model.setInputSize(320, 320)
    model.setInputScale(1.0/127.5)
    model.setInputMean((127.5, 127.5, 127.5))
    model.setInputSwapRB(True)
    
    cam = cv2.VideoCapture(cam_idx)
    while True:
        success, img = cam.read()
        if not success:
            break
        # 执行检测并处理结果
        classes, confs, boxes = model.detect(img, confThreshold=0.5)
        if len(classes) > 0:
            for classId, confidence, box in zip(classes.flatten(), confs.flatten(), boxes):
                cv2.rectangle(img, box, (0,255,0), 2)
                cv2.putText(img, f"{classId}: {confidence:.2f}", 
                            (box[0]+10, box[1]+30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,255,0), 2)
        cv2.imshow(window_name, img)
        if cv2.waitKey(2) & 0xFF == ord('x'):
            break
    cam.release()
    cv2.destroyWindow(window_name)

# 启动双线程检测
thread1 = threading.Thread(target=detect_camera, args=(0, "Camera 0"))
thread2 = threading.Thread(target=detect_camera, args=(1, "Camera 1"))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

cv2.destroyAllWindows()

方案2：单线程读帧+队列调度检测 #

主线程负责读取两个摄像头的帧并放入队列，单独用一个线程处理检测，避免多线程操作模型：

import cv2
import threading
from queue import Queue

def detect_worker(model, input_queue, output_queue):
    while True:
        frame_data = input_queue.get()
        if frame_data is None:  # 终止信号
            break
        cam_idx, img = frame_data
        # 统一执行检测
        classes, confs, boxes = model.detect(img, confThreshold=0.5)
        # 绘制检测结果
        if len(classes) > 0:
            for classId, confidence, box in zip(classes.flatten(), confs.flatten(), boxes):
                cv2.rectangle(img, box, (0,255,0), 2)
                cv2.putText(img, f"{classId}: {confidence:.2f}", 
                            (box[0]+10, box[1]+30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,255,0), 2)
        output_queue.put((cam_idx, img))
        input_queue.task_done()

def main():
    # 初始化单个模型实例
    model = cv2.dnn_DetectionModel(weightsPath, configPath)
    model.setInputSize(320, 320)
    model.setInputScale(1.0/127.5)
    model.setInputMean((127.5, 127.5, 127.5))
    model.setInputSwapRB(True)

    input_queue = Queue(maxsize=10)
    output_queue = Queue(maxsize=10)

    # 启动检测线程
    worker_thread = threading.Thread(target=detect_worker, args=(model, input_queue, output_queue))
    worker_thread.daemon = True
    worker_thread.start()

    cam0 = cv2.VideoCapture(0)
    cam1 = cv2.VideoCapture(1)

    while True:
        # 读取双摄像头帧
        success0, img0 = cam0.read()
        success1, img1 = cam1.read()

        if success0:
            input_queue.put((0, img0.copy()))
        if success1:
            input_queue.put((1, img1.copy()))

        # 显示处理后的帧
        while not output_queue.empty():
            cam_idx, processed_img = output_queue.get()
            cv2.imshow(f"Camera {cam_idx}", processed_img)
            output_queue.task_done()

        if cv2.waitKey(2) & 0xFF == ord('x'):
            break

    input_queue.put(None)  # 发送终止信号
    worker_thread.join()

    cam0.release()
    cam1.release()
    cv2.destroyAllWindows()

if __name__ == "__main__":
    main()

方案3：线程锁保护模型调用 #

若受限于内存无法创建多个模型实例，可使用线程锁确保同一时间只有一个线程调用检测方法：

import cv2
import threading

# 初始化单个模型实例
model = cv2.dnn_DetectionModel(weightsPath, configPath)
model.setInputSize(320, 320)
model.setInputScale(1.0/127.5)
model.setInputMean((127.5, 127.5, 127.5))
model.setInputSwapRB(True)

detect_lock = threading.Lock()  # 创建线程锁

def detect_camera(cam_idx, window_name):
    cam = cv2.VideoCapture(cam_idx)
    while True:
        success, img = cam.read()
        if not success:
            break
        # 加锁确保模型检测串行执行
        with detect_lock:
            classes, confs, boxes = model.detect(img, confThreshold=0.5)
        # 绘制检测结果
        if len(classes) > 0:
            for classId, confidence, box in zip(classes.flatten(), confs.flatten(), boxes):
                cv2.rectangle(img, box, (0,255,0), 2)
                cv2.putText(img, f"{classId}: {confidence:.2f}", 
                            (box[0]+10, box[1]+30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,255,0), 2)
        cv2.imshow(window_name, img)
        if cv2.waitKey(2) & 0xFF == ord('x'):
            break
    cam.release()
    cv2.destroyWindow(window_name)

# 启动双线程
thread1 = threading.Thread(target=detect_camera, args=(0, "Camera 0"))
thread2 = threading.Thread(target=detect_camera, args=(1, "Camera 1"))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

cv2.destroyAllWindows()

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