视频监控系统已经不只是录像回放工具。当AI能力加入后，摄像头采集的每一帧都可能触发人员闯入、烟雾火焰、车辆违停等事件。要稳定处理一路或多路视频，系统需要同时解决流媒体接入、AI推理、告警分发三件事，任何一个环节出现阻塞都会直接影响实时性。
流媒体接入与解码：先把视频稳定地变成帧
摄像头厂商普遍支持RTSP，部分新设备支持GB28181、RTMP或WebRTC。以RTSP为例，拉流时通常使用FFmpeg或GStreamer。FFmpeg支持TCP和UDP传输，TCP更适合公网或弱网环境，因为丢包率低；UDP虽然延迟略低，但容易出现花屏。实时监控建议显式指定TCP并将探测时间调短，避免拉流阶段卡住。下面是一个用FFmpeg拉取RTSP流并输出H.264裸流的命令示例。
ffmpeg -rtsp_transport tcp -i rtsp://user:pass@192.168.1.64:554/stream1 \ -an -vcodec copy -f h264 output.h264
解码阶段不要直接使用CPU软解多路1080P视频，否则会迅速耗尽资源。正确做法是使用NVIDIA NVDEC、Intel QSV或Rockchip MPP等硬件解码。如果使用Python，可以借助PyAV或OpenCV的VideoCapture，但要注意它们默认可能使用软解。GStreamer的nvv4l2decoder和FFmpeg的h264_cuvid都能显著降低CPU占用。另一个关键点是GOP长度，如果关键帧间隔过大，断线重连后可能长时间无法出图。很多场景将GOP设置为1到2秒，牺牲少量压缩率换取更快的恢复速度。
对于多路视频，建议每路拉流和解码独立运行，并通过队列将解码后的帧交给推理模块，避免某一路卡顿拖垮整个服务。解码线程只负责取帧、缩放到模型输入尺寸、放入队列；推理线程从队列批量取出帧，拼成batch再调用模型。这里需要注意队列长度控制，如果推理跟不上就丢弃旧帧，保证告警延迟不会被积压不断放大。
AI推理与告警判定：用异步队列替代同步阻塞
如果把视频帧直接送入模型做同步推理，每帧耗时30毫秒，一路25FPS视频每秒就需要750毫秒GPU时间，已经无法跟上实时节奏。更合理的做法是抽帧检测，例如每秒检测5到8帧，同时依赖跟踪算法补齐中间帧的状态。这样既能降低算力消耗，又能保持事件触发的时间精度。
抽帧之后进入异步队列。以下是一个简化的Python示例，展示帧队列和推理循环的基本结构。解码线程负责读取视频帧并缩放，推理线程按小批量方式调用模型，避免单帧推理带来的GPU利用率不足。
import queue
import cv2
frame_queue = queue.Queue(maxsize=32)
def decode_loop(cap):
while True:
ret, frame = cap.read()
if not ret:
break
resized = cv2.resize(frame, (640, 384))
if frame_queue.qsize() >= 30:
continue
frame_queue.put(resized)
def infer_loop(model):
while True:
batch = []
for _ in range(4):
if frame_queue.empty():
break
batch.append(frame_queue.get())
if not batch:
continue
results = model(batch)
handle_results(results)
告警判定不应该只看单帧结果。比如人员闯入检测，单帧可能因为遮挡或误检产生虚警。常见做法是连续N帧或时间窗内出现目标超过阈值才触发告警，同时对同一目标ID做去重。多目标跟踪算法如ByteTrack、DeepSORT可以根据检测框做关联，每个目标有稳定ID。告警触发时可以携带目标位置、置信度、连续出现时长、截图或短视频片段。这样下游业务才能做进一步复核。将判定逻辑放在推理线程之后、推送之前，形成一条清晰的流水线。
告警风暴抑制：冷却、聚合与优先级
一旦系统上线，最容易出现的问题不是漏报，而是告警风暴。一个打伞的人可能被识别为人员和异物，同一事件在几秒内触发几十次，运维人员很快会忽略所有通知。抑制告警需要从时间窗口、目标维度、空间区域三个层面做合并。
冷却机制可以按事件类型、摄像头ID和目标ID组合生成唯一键，在冷却时间内相同键的告警只保留第一条或合并计数。比如发现某摄像头区域有人闯入，第一次推送后进入30秒冷却，期间继续出现只增加内部计数，冷却结束后如果事件仍存在再发一次升级提醒。下面这个Python类实现了基于时间戳的简单冷却判断。
class AlertCooldown:
def __init__(self, cooldown_seconds):
self.cooldown = cooldown_seconds
self.last_sent = {}
def should_send(self, key, now):
last = self.last_sent.get(key)
if last is None or now - last > self.cooldown:
self.last_sent[key] = now
return True
return False
优先级用于重要事件优先送达。例如火焰检测、安全帽缺失、区域入侵的优先级不同。可以为告警定义level字段，将火焰检测设为最高级，区域入侵次之，安全帽缺失再次之。高优先级事件可绕过部分冷却限制，甚至走短信或电话通道；低优先级仅写数据库或聚合日报。聚合时还可以按摄像头、时间段汇总，比如某园区东门过去5分钟产生12次人员靠近周界告警，比一条条推送更实用。
性能优化与落地细节
多路实时监控的性能瓶颈往往不在模型本身，而在数据搬运和内存分配。解码后的YUV帧如果反复拷贝到CPU内存再上传GPU，PCIe带宽可能先成为瓶颈。理想情况下，解码、缩放、推理都在GPU上完成，减少不必要的拷贝。NVIDIA DeepStream、华为MindX等框架封装了这种pipeline，适合需要快速上线的团队。
还要关注时钟同步。多路视频的时间戳可能来自摄像头本地时钟，不同摄像头之间偏差可能达到几秒甚至几分钟。如果做跨摄像头目标追踪或事件关联，必须统一到同一时钟源。可以在接入层用NTP校准，或使用流媒体服务器写入统一的UTC时间戳。对于断流，需要设置超时重连和指数退避，避免反复拉起进程。可以用如下伪代码描述重连策略。
while True:
try:
connect_and_read()
except StreamError:
time.sleep(min(backoff, 60))
backoff *= 2
else:
backoff = 1
最后是压测和容量规划。一路1080P H.264码流约4Mbps，千兆网口理论上可以承载200路以上，但CPU、GPU解码和推理能力才是限制。建议先用小规模环境测出单卡能稳定支持的并发路数，再按N+1冗余部署。告警存储和截图的磁盘增长速度也要提前评估，避免因磁盘写满导致服务崩溃。整个系统从拉流、抽帧、推理、告警到推送，尽可能解耦，任何一环出现异常都不能影响其他环节。