如何用Python+YOLOv4实现50路摄像头并行实时目标检测？

核心问题说明 #

你之前多线程、多进程方案失败是两个典型坑：

• Python多线程受GIL全局解释器锁限制，model.detect()这类计算密集型任务无法通过多线程实现真正并行，本质还是串行执行
• 多进程场景下如果在主进程提前加载模型再传入子进程，会因为OpenCV DNN对象不支持跨进程序列化、内部状态锁冲突导致进程崩溃、卡死，必须让每个工作子进程独立初始化自己的模型实例，不能共享主进程的模型对象

可直接运行的并行验证代码 #

以下代码用Python标准库multiprocessing实现进程池，每个子进程启动时仅加载一次模型，后续并行处理分配到的图像任务，完全符合无串行的并行要求，测试阶段直接对接你的50张测试图像即可，后续替换为摄像头流逻辑不需要改核心并行框架：

import os
import time
import cv2
import multiprocessing

# 子进程全局变量，每个进程独立持有自己的模型实例
subprocess_model = None
coco_classes = ["car", "plate", "motorcycle"]

def init_subprocess_model():
    """子进程初始化函数：每个进程启动时执行一次，加载自己的模型，避免重复加载开销"""
    global subprocess_model
    net = cv2.dnn.readNet("custom.weights", "custom.cfg")
    # 如果你编译了带CUDA支持的OpenCV，可以打开下面两行用GPU加速，默认用CPU
    # net.setPreferableBackend(cv2.dnn.DNN_BACKEND_CUDA)
    # net.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA)
    subprocess_model = cv2.dnn_DetectionModel(net)
    subprocess_model.setInputParams(size=(416, 416), scale=1 / 255, swapRB=True)

def detect_single_image(fn):
    """单张图像检测函数，每个子进程调用该函数处理分配到的任务"""
    global subprocess_model
    img_path = os.path.join('./img/', fn)
    image = cv2.imread(img_path)
    if image is None:
        return f"Read image failed: {img_path}"
    
    t = time.time()
    c, v, b = subprocess_model.detect(image, 0.2, 0.4)
    cost = int((time.time() - t) * 1000)
    c = [coco_classes[x] for x in c]
    return f"{cost}ms : {list(zip(c, v))}"

if __name__ == '__main__':
    # 进程数配置：CPU推理设为和CPU物理核心数一致即可，GPU推理根据显存大小调整，不要盲目开50个避免资源不足
    WORKER_PROCESS_NUM = min(multiprocessing.cpu_count(), 50)
    img_path = './img/'
    file_list = [fn for fn in os.listdir(img_path) if fn.lower().endswith(('.jpg', '.jpeg', '.png', '.bmp'))]

    print(f"Starting parallel detection with {WORKER_PROCESS_NUM} processes, total {len(file_list)} images")
    start_time = time.time()
    # 初始化进程池，每个进程启动时执行init_subprocess_model加载模型
    with multiprocessing.Pool(processes=WORKER_PROCESS_NUM, initializer=init_subprocess_model) as pool:
        # 并行分发所有检测任务
        results = pool.map(detect_single_image, file_list)
    total_cost = time.time() - start_time
    print(f"All detection finished, total cost: {total_cost:.2f}s")
    # 统一打印结果，避免多进程同时print导致输出错乱
    for res in results:
        print(res)

后续对接真实摄像头的调整说明 #

测试验证通过后要接50路真实摄像头，只需要做两处调整：

• 把file_list替换成50路摄像头的RTSP地址/设备ID列表
• 把detect_single_image里读静态图片的逻辑换成循环读取对应摄像头的帧、执行检测的逻辑即可，每个进程固定处理一路摄像头，完全并行无串行

注意事项 #

• 不要直接把进程数设为50：CPU推理时进程数超过物理核心数会因为CPU频繁上下文切换反而降低速度；GPU推理时单张显卡显存如果不足以加载对应数量的模型实例，会触发显存溢出报错，根据硬件情况调整WORKER_PROCESS_NUM即可
• Windows系统运行时必须保留if __name__ == '__main__':的判断，否则多进程会递归启动导致报错
• 如果需要更高的吞吐量，可以后续调整为批量推理逻辑，但当前代码已经满足并行处理、无串行逻辑的要求

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