YOLOv8多摄像头接入实战：并发检测系统搭建-平芜编程栈

YOLOv8多摄像头接入实战：并发检测系统搭建

1. 引言

1.1 业务场景描述

在智能安防、工业质检、交通监控等实际应用中，单一摄像头的视野受限，难以满足大范围、多角度实时监控的需求。为了实现对复杂场景的全面感知，多摄像头并发目标检测系统成为关键基础设施。本文将基于Ultralytics YOLOv8 轻量级模型（v8n），构建一个支持多路视频流并行处理的工业级目标检测系统，集成可视化 WebUI 与实时统计看板，适用于 CPU 环境下的低成本部署。

1.2 痛点分析

传统单摄像头检测方案存在以下问题：

覆盖盲区：无法实现全方位监控
数据孤岛：各摄像头独立运行，缺乏统一统计与联动
性能瓶颈：多路视频串行处理导致延迟高、吞吐低
部署成本高：依赖 GPU 推理，难以在边缘设备普及

而现有开源项目大多仅支持单路输入或需依赖 ModelScope 等平台模型，灵活性差且稳定性不足。

1.3 方案预告

本文提出一种基于Flask + OpenCV + Ultralytics YOLOv8n的轻量化多摄像头并发检测架构，具备以下特性：

支持N 路 RTSP/USB/IP 摄像头同时接入
多线程异步推理，避免阻塞
内置 Web 可视化界面，展示每路画面及全局物体统计
针对 CPU 进行优化，单帧推理时间控制在毫秒级
完全脱离 ModelScope，使用官方 Ultralytics 引擎，稳定零报错

2. 技术方案选型

2.1 核心组件对比

组件	候选方案	选择理由
目标检测模型	YOLOv5 / YOLOv8 / SSD / Faster R-CNN	选用YOLOv8n：速度更快、小目标召回率更高、API 更简洁
视频采集库	OpenCV / GStreamer / FFmpeg	选用OpenCV：易用性强，兼容 USB/RTSP/IP 摄像头
后端框架	Flask / FastAPI / Django	选用Flask：轻量、适合嵌入式部署，易于集成 OpenCV
并发模式	多线程 / 多进程 / asyncio	选用多线程 + 队列缓冲：降低资源开销，避免 GIL 影响
前端展示	Streamlit / Vue.js / 原生 HTML+JS	选用原生 HTML+JS+CSS：减少依赖，提升加载速度

2.2 为什么选择 YOLOv8？

YOLOv8 是 Ultralytics 团队推出的最新一代目标检测模型，在保持高精度的同时大幅优化了推理速度。其Nano 版本（v8n）尤其适合 CPU 部署：

参数量仅约 300 万，模型大小 < 5MB
在 Intel i5 CPU 上推理速度可达30~50ms/帧
支持 COCO 数据集 80 类常见物体识别（人、车、动物、家具等）
提供.predict()接口，一行代码完成推理

results = model.predict(frame, conf=0.4)

此外，YOLOv8 原生支持数量统计功能，可通过results[0].boxes.cls获取类别索引，结合Counter快速生成报告。

3. 实现步骤详解

3.1 环境准备

# 创建虚拟环境 python -m venv yolov8_env source yolov8_env/bin/activate # Linux/Mac # yolov8_env\Scripts\activate # Windows # 安装核心依赖 pip install ultralytics opencv-python flask numpy

⚠️ 注意：建议使用 Python 3.8~3.10，避免与 PyTorch 兼容性问题。

3.2 模型初始化与加载

from ultralytics import YOLO import cv2 import threading import time from collections import Counter # 加载 YOLOv8n 模型（CPU 模式） model = YOLO('yolov8n.pt') # 自动下载预训练权重 # 全局变量：存储每路摄像头的状态和结果 cameras = {} stats_lock = threading.Lock() global_stats = Counter()

3.3 多摄像头采集线程设计

为每一路摄像头创建独立采集线程，使用cv2.VideoCapture读取视频流，并通过队列缓存最新帧以防止内存溢出。

import queue def camera_capture_thread(cam_id, url): cap = cv2.VideoCapture(url) if not cap.isOpened(): print(f"[ERROR] 无法打开摄像头 {cam_id}") return frame_queue = queue.Queue(maxsize=2) # 缓冲最近两帧 cameras[cam_id] = {'queue': frame_queue, 'last_result': None} while True: ret, frame = cap.read() if not ret: time.sleep(1) continue # 缩放图像以加快推理速度 h, w = frame.shape[:2] scale = 640 / max(h, w) new_h, new_w = int(h * scale), int(w * scale) resized = cv2.resize(frame, (new_w, new_h)) # 存入队列（自动丢弃旧帧） if not frame_queue.full(): try: frame_queue.put_nowait(resized) except queue.Full: pass time.sleep(0.01) # 控制采集频率

3.4 并发推理线程实现

每个摄像头对应一个推理线程，从队列中取出最新帧进行 YOLOv8 推理，并更新本地结果与全局统计。

def inference_thread(cam_id): global global_stats frame_queue = cameras[cam_id]['queue'] while True: try: frame = frame_queue.get(timeout=1) except queue.Empty: continue # 使用 YOLOv8 进行推理 results = model(frame, verbose=False) annotated_frame = results[0].plot() # 绘制检测框 # 提取类别名称用于统计 names_dict = model.model.names cls_tensor = results[0].boxes.cls class_ids = cls_tensor.cpu().numpy().astype(int) class_names = [names_dict[id] for id in class_ids] count = Counter(class_names) # 更新本地结果 with stats_lock: cameras[cam_id]['last_result'] = (annotated_frame, count) # 合并到全局统计 global_stats.clear() for c in [cam['last_result'][1] for cam in cameras.values() if cam['last_result']]: global_stats += c time.sleep(0.01)

3.5 Web 可视化服务搭建（Flask）

提供 HTTP 接口，返回合并后的多路画面拼接图与 JSON 统计数据。

from flask import Flask, Response, jsonify import numpy as np app = Flask(__name__) @app.route('/video_feed') def video_feed(): def generate(): while True: frames_to_show = [] for cam_id in sorted(cameras.keys()): data = cameras[cam_id].get('last_result') if data: frame, _ = data frames_to_show.append(frame) if frames_to_show: # 拼接多路画面（最多4路） if len(frames_to_show) == 1: combined = frames_to_show[0] elif len(frames_to_show) <= 2: combined = np.hstack(frames_to_show) else: h = frames_to_show[0].shape[0] row1 = np.hstack(frames_to_show[:2]) row2 = np.hstack(frames_to_show[2:4]) combined = np.vstack([row1, row2]) _, buffer = cv2.imencode('.jpg', combined, [cv2.IMWRITE_JPEG_QUALITY, 85]) yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n') else: time.sleep(0.1) return Response(generate(), mimetype='multipart/x-mixed-replace; boundary=frame') @app.route('/stats') def get_stats(): with stats_lock: return jsonify(dict(global_stats))

3.6 主程序启动逻辑

if __name__ == '__main__': # 添加摄像头（示例：本地摄像头 + RTSP 流） urls = { 0: 0, # 本地摄像头 1: 'rtsp://admin:password@192.168.1.100:554/stream1', 2: 'rtsp://admin:password@192.168.1.101:554/stream1' } # 启动采集线程 for cam_id, url in urls.items(): t = threading.Thread(target=camera_capture_thread, args=(cam_id, url), daemon=True) t.start() # 启动推理线程 for cam_id in urls.keys(): t = threading.Thread(target=inference_thread, args=(cam_id,), daemon=True) t.start() # 启动 Flask 服务 app.run(host='0.0.0.0', port=5000, threaded=True)

4. 实践问题与优化

4.1 常见问题及解决方案

问题	原因	解决方法
视频卡顿、延迟高	单线程处理多路流	改为多线程采集+推理
内存占用过高	帧堆积未释放	使用`queue.Queue(maxsize=2)`限制缓存
检测结果闪烁	不同帧间类别抖动	增加置信度阈值（`conf=0.4`）或滑动平均
Flask 页面无法访问	默认绑定 localhost	设置`host='0.0.0.0'`开放外网访问

4.2 性能优化建议

图像预缩放：将输入分辨率统一调整至 640px 最长边，显著降低计算量
跳帧推理：非关键场景可设置每 2~3 帧推理一次，提升吞吐
模型量化：导出为 ONNX 或 TensorRT 格式，进一步加速 CPU 推理
静态HTML缓存：前端页面使用缓存策略，减少重复加载

5. 总结

5.1 实践经验总结

本文实现了基于 YOLOv8 的多摄像头并发检测系统，具备以下核心价值：

✅工业级稳定性：采用官方 Ultralytics 引擎，避免第三方平台依赖
✅真正的并发处理：多线程架构保障 N 路摄像头无阻塞运行
✅轻量化部署：YOLOv8n 模型可在普通 CPU 设备上实现实时检测
✅即开即用：集成 WebUI 展示画面与统计看板，无需额外配置

5.2 最佳实践建议

摄像头命名规范：为每路流分配唯一 ID 和标签（如“入口”、“车间A”），便于管理
定期重启机制：长时间运行可能导致 OpenCV 资源泄漏，建议每日重启服务
日志记录增强：添加异常捕获与日志输出，便于故障排查

获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

YOLOv8多摄像头接入实战：并发检测系统搭建