1. 服务层设计：先想清楚架构再写代码

写这一篇之前，先把前面的路梳理一下。RK3588 上跑 YOLOv5s，模型转换、RKNN 推理、单张图片的检测，这些在第零篇到第六篇里已经折腾得差不多了。但一个检测功能要真正变成能给别人调的东西，必须有一套服务层把它包起来，还要把摄像头这个数据源接进来。我做这套东西的定位就是一个玩具级的边缘智能盒子，摄像头对着办公区域，后端提供 HTTP 接口给内网的小页面调用。如果你是做巡检设备、通道监控、或者无人值守柜机这类场景，这篇的思路可以直接参考。

服务层这块，我见过不少人上来就直接在板子上写一个普通脚本循环推帧，然后前端直接连摄像头裸流。这样demo阶段确实能弹出一个窗口看到检测框，但你要给别人用、要跟别的系统对接，就完全行不通。原因很直接：第一，摄像头裸流的协议和地址属于底层细节，不应该暴露给调用方；第二，检测结果的输出需要一个统一的数据结构，不能各写各的；第三，模型初始化和推理状态必须常驻内存，不能让每次请求都重新加载一遍模型，那样延迟和抖动会大到你怀疑人生。所以我的做法是：把 RKNN 模型、摄像头取流、后处理逻辑全部封装在一个 Python 服务进程里，对外只开一套简单的 HTTP 接口。

服务框架上我选了 Flask，而不是 FastAPI 或者其他更重的方案。理由很朴素：板子上的资源紧张，Python 服务本身就不该承担太多额外依赖，Flask 体积小、生态成熟，最重要的是我对它的线程模型和部署方式熟悉，出问题好排查。FastAPI 确实性能更好，异步特性在 IO 密集场景有优势，但我们的核心瓶颈在推理和视频解码，不在 HTTP 传输环节，框架层面的那点差异根本体现不出来。gRPC 我也考虑过，但给内网小工具引入 protobuf 编译链和代码生成，纯属给自己找麻烦。

模型加载策略这件事，我在第五篇里其实提到过，但服务层这里要再强调一次：进程启动时初始化一次 RKNN 模型，全局持有一个对象，后续所有请求共用它。这里有个很关键的点：RK3588 的 NPU 是单路径的，多线程同时调rknn.inference()会互相干扰，所以服务层必须把推理动作串行化。我的做法是一个threading.Lock()，每个请求进来先抢锁，抢不到就等着，等到的那个请求发布推理指令。这样虽然并发能力不高，但保证每个请求的响应是稳定的，不会出现两个线程同时写 NPU 寄存器导致推理结果错乱的情况。实测下来单路摄像头、单请求吞吐的场景，这套模型已经足够。

接口设计上，我只给了两个业务接口：一个POST /detect用来上传图片检测，一个POST /video用来做视频流检测并返回最新一帧的标注结果。检测接口的输入是三样东西：图片的 base64 字符串、摄像头标识、帧序号；输出是检测框列表、类别、置信度、当前帧推理耗时。返回格式统一是 JSON，因为下游不管是网页前端还是小程序，解析 JSON 都是最省事的。

部署方式上，Flask 自带的开发服务器千万不能直接用于板端服务，那个werkzeug内置 server 是单进程的，而且性能稀烂。我在板子上用的是waitress，纯 Python 实现的 WSGI 服务器，没有 C 扩展，在 ARM 上安装不会出编译问题，单进程多线程的模式也匹配我们的串行推理设计。启动命令就三行，比 gunicorn 省心很多，因为在 ARM 上 gunicorn 有时会因为缺少gevent或eventlet导致并发模型不生效，反而掩盖了真正的线程问题。

pip install waitress waitress-serve --host=0.0.0.0 --port=8080 app:app

2. 摄像头接入：从 RTSP 地址到稳定的视频帧

摄像头接入是我这一路下来最不想提又必须详细写的一段，因为坑太密了。先说最基础的 RTSP 取流地址。市面上的摄像头，海康的地址格式一般是rtsp://用户名:密码@IP:554/Streaming/Channels/101，后面三位数字含义是通道号、码流类型、协议类型，101 表示通道一的主码流，102 表示通道一的子码流。宇视的一般是rtsp://用户名:密码@IP:554/video1或/video2。杂牌摄像头最头疼，有的用 ONVIF 标准可以自动探测，有的就只能靠厂家给的私有格式。如果你手上是一台来路不明的杂牌摄像头，我建议先用 ONVIF Device Manager 或类似的工具扫一下设备，拿到取流地址再写进代码。

关于码流选择，我直接给结论：日常调试和预览用子码流，分辨率低、帧率高、带宽占用小；真正要交给检测模型推理的，建议也先用子码流。为什么？YOLOv5s 的输入分辨率普遍是 640x640，你送 1080P 的原图和送 960x540 的图进去，检测框的坐标最终都要映射回原图尺寸，模型的检测能力并不会因为你送了高清图就大幅提升，但高清主码流对解码性能和网络带宽的压力是成倍增加的。我在板子上测试，同一路摄像头，主码流时 CPU 占用比子码流高出一大截，而检测效果几乎没有肉眼可见的区别。所以做实时检测的嵌入设备，最优解是限制摄像头输出或主动解码到较低分辨率，而不是盲目追求高画质。

取流程序的选型，我建议直接避开裸 OpenCV 的cv2.VideoCapture("rtsp://...")。不是说完全不能用，而是它默认走的 FFmpeg 后端在 RK3588 上表现很不稳定。具体问题后面细说，这里先讲推荐方案：用 GStreamer 管道。GStreamer 在 Linux 嵌入平台上的适配度远高于 OpenCV 自带的解码链路，而且能直接调用 RK3588 的 VPU 硬解 H.264/H.265，CPU 占用能压低一个量级。

我的取流管道长这样：

gst-launch-1.0 rtspsrc location=rtsp://user:pass@192.168.1.64:554/Streaming/Channels/102 \ latency=0 \ rtph264depay ! h264parse ! v4l2h264dec ! videoconvert ! \ video/x-raw,format=RGB,width=640,height=640,framerate=30/1 ! appsink

这条管道做了几件事：rtspsrc负责 RTSP 协议解析，latency=0强制低延迟；rtph264depay把 RTP 包还原成 H.264 裸流；h264parse做码流解析；v4l2h264dec调用 Video4Linux2 的硬件解码器，这就是 RK3588 的 VPU 在工作；videoconvert做颜色空间转换；最后appsink把解码后的图像帧交给 Python 程序。Python 端只需要从appsink的缓冲队列里取 fames，格式已经是 RGB 的 numpy 数组，可以直接喂给后面的预处理。

import cv2 pipeline = ( "rtspsrc location=rtsp://user:pass@192.168.1.64:554/Streaming/Channels/102 " "latency=0 ! rtph264depay ! h264parse ! v4l2h264dec ! " "videoconvert ! video/x-raw,format=RGB,width=640,height=640,framerate=30/1 ! " "appsink drop=1" ) cap = cv2.VideoCapture(pipeline, cv2.CAP_GSTREAMER)

注意cv2.VideoCapture的第一个参数可以是 GStreamer 管道字符串，但前提是 OpenCV 编译时带着 GStreamer 支持。你自己用源码编 OpenCV，编译参数里要加-DWITH_GSTREAMER=ON；如果你直接用pip install opencv-python，那后面挂cv2.CAP_GSTREAMER大概率报错，因为官方 pip 包默认不带 GStreamer 后端。在板子上我最后是自己编的 OpenCV，这一步在第六篇的依赖准备里写过，这里不重复了。

如果实在不想折腾 GStreamer，还有一个兜底方案：用ffmpeg子进程拉流，通过标准输出管道把原始帧送给 OpenCV。这个方案虽然多了一层进程间通信，但胜在逻辑简单、不依赖 OpenCV 的编译配置。命令大概是：

ffmpeg -rtsp_transport tcp -i rtsp://user:pass@ip:554/Streaming/Channels/102 \ -f rawvideo -pix_fmt bgr24 -s 640x640 -an pipe:1

Python 端用subprocess.Popen启动这个命令，然后循环从proc.stdout读取640 * 640 * 3字节的数据，重组成 numpy 数组。这套方案的特点是稳，ffmpeg 对 RTSP 的兼容性是出了名的好，但也因此牺牲了硬解能力，CPU 占用会偏高。如果你只是做验证性测试，这个兜底方案足够；如果是长期稳定运行的项目，我还是推荐把 GStreamer 调通，这样整个系统才能有空余算力去跑模型推理。

3. 推理链路实现：把一帧画面变成检测结果

摄像头把视频帧送进来之后，就要开始真正的推理链路了。这里有一个特别多新手踩的坑：格式对齐。RKNN 模型训练和转换时，输入张量约定的是 RGB 顺序、CHW 布局、fp32 归一化数据；但 OpenCV 和 GStreamer 默认给出来的是 BGR 顺序、HWC 布局、uint8 原始值。这两者对不上，直接把数据塞进rknn.inference()，出来的检测结果就是一团乱。你可能怎么调参数都调不对，最后怀疑模型没转换好，其实只是颜色通道顺序的问题。

格式转换的代码每一版项目我都要重新写一遍，干脆贴出来供你直接抄：

def preprocess(frame_rgb): # frame_rgb: uint8, shape=(H, W, 3), RGB order from GStreamer # RKNN expects NCHW, RGB, fp32, [0, 1] img = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2RGB) # 占位，直观表示通道顺序 img = img.transpose(2, 0, 1) # HWC -> CHW img = np.ascontiguousarray(img) # 转成内存连续，避免后面报错 img = img.astype(np.float32) / 255.0 # uint8 -> fp32 归一化 img = np.expand_dims(img, axis=0) # CHW -> NCHW return img

这里有几个容易被忽略的细节：transpose之后得到的数组是原数据的一个视图，内存布局不是连续的，直接塞给 RKNN 的 C 扩展接口有时候会报ValueError: ndarray is not contiguous，所以必须紧跟一个np.ascontiguousarray。astype之后最好再确认一下dtype是不是float32，因为有些版本的 RKNN Python 接口对 float64 输入不报错但结果全错，那真是白排查一整天。

预处理之后进推理。RKNN 的调用很简单：

outputs = rknn.inference(inputs=[img_nchw], data_format="nchw")

data_format这个参数很关键。如果你的输入已经是 NCHW 布局，这里就写"nchw"；如果你图省事不做transpose，直接送 HWC 数据，那这里要写"nhwc"，而且通道顺序还得再调整。我建议统一走 NCHW，因为 ONNX 转出来的 RKNN 模型内部就是这个布局，少一层隐式转换就少一个可能的坑。

推理出来的outputs是三个特征图的列表，分别对应 YOLOv5s 的 80x80、40x40、20x20 三个尺度输出。这一步不能直接拿来当检测框用，得先做一次 NMS 后处理。我用的还是 YOLOv5 仓库里的那一套 anchors 和类别表，后处理逻辑是：先按置信度阈值筛掉低分框，然后用 sigmoid 激活得到目标概率，再按 anchor 策略计算框的坐标，最后对所有类别做 NMS 抑制重叠框。这部分代码不复杂，但如果你从头自己写，很费时间，不如直接从 ultralytics 的官方代码里抄一个简化版，把 anchors、strides、class names 配置好就行。

后处理还有一个容易忽略的点：特征图上的坐标是缩放过的，要映射回原图必须乘上对应的 stride。比如 80x80 特征图的 stride 是 8，40x40 的是 16，20x20 的是 32。坐标映射之后，过一遍np.clip把框限制在图像边界内，再转成 JSON 里要用的整数。这一步不做好，画出来的检测框会飘出画面外，或者歪到离谱的位置。

关于帧缓冲，我强烈建议用“只保留最新帧”的策略，而不是无脑排队。做法是一个queue.Queue(maxsize=1)，取流线程往队列里放帧时用put_nowait，如果队列里已经有旧帧就把它丢弃再放新帧。为什么要这样设计？因为模型推理的时间是固定的，假设一帧推理要 30ms，摄像头是 30fps，也就是每帧间隔 33ms，看起来刚好能跟上。但如果你遇到某帧推理突然慢一点，或者系统忙了一下，队列里就会积压帧。这时候如果按顺序处理旧帧，实时性就废了——你输出的检测结果永远是几百毫秒前的画面，这对监控类应用是没有意义的。所以必须强制只处理最新帧，保证延迟始终维持在一个低水位。

frame_queue = queue.Queue(maxsize=1) def capture_loop(): while True: ret, frame = cap.read() if not ret: continue if frame_queue.full(): try: frame_queue.get_nowait() # 丢旧帧 except queue.Empty: pass frame_queue.put_nowait(frame) def inference_loop(): while True: frame = frame_queue.get() result = do_detect(frame) latest_result = result

4. 那个折腾最久的坑：GStreamer 管道里的花屏、卡顿和神秘崩溃

这个系列写到现在，终于要说到标题里那个“折腾最久的坑”了。问题说起来很简单，但排查过程极其痛苦，我希望能帮你少走几周弯路。

当时的情况是这样的：我已经把 GStreamer 管道调通，取流、推理都各自独立验证过，单帧检测也没问题。但把整个服务跑起来，让它连续运行几分钟后，问题就出现了——画面开始花屏，表现为大面积的绿色色块和条纹闪烁，过一会儿整个画面直接冻结，然后进程崩溃，日志里连个像样的报错都没有，只有一行类似于Segmentation fault的信息。最初我以为摄像头坏了，因为我用的是一台杂牌摄像头，取流地址也是用工具扫出来的，本身兼容性就让人心里打鼓。我换了一台海康的摄像头试，结果问题依旧，只是出现的频率低了一些。这个时候我确定，问题大概率不在摄像头本身，而是在我的取流和解码链路上。

接着我开始怀疑网络。我把摄像头的网线直接从交换机上拔下来，用一根短网线直连板子和摄像头，问题照样出现。理论上局域网内直连不会有丢包，所以网络也暂时排除了。那问题就只能在 GStreamer 这条管道里了。我知道 GStreamer 的管道如果某些插件版本不匹配，或者 caps 协商失败，是会发生一些诡异现象的，但通常会在终端里打印警告，不至于直接段错误。

于是我开始做对照实验：我用gst-launch-1.0在命令行里播放同一个 RTSP 流，连续跑了一个小时，一次都没崩溃。这说明 GStreamer 本身和系统的 VPU 驱动是正常的。我又用 Python 的 GStreamer appsink 手动一帧一帧读，也没有崩溃。最后我把 OpenCV 的cap.read()调用换成cap.grab()加cap.retrieve()分步操作，发现问题开始复现了。直到我把 OpenCV 的getBuildInformation()打印出来仔细看，才意识到问题所在：我用的这个 OpenCV 虽然在编译时开了 GStreamer，但它接的是 GStreamer 的appsink，而appsink的缓冲机制和 OpenCV 内部的retrieve逻辑在某些版本上存在一个很隐蔽的竞态，导致缓冲区被重复释放或错位引用。当 CPU 负载升高（推理线程和取流线程同时跑），竞态发生概率急剧上升，最终表现为花屏和崩溃。

这个问题折腾了我整整一个周末。最终的修复方案分两步。第一步，换掉appsink的drop=1属性，在管道的appsink前面加上一个queue max-size-buffers=2 leaky=downstream。这个leaky=downstream的作用是当缓冲满了时丢弃新帧，同时通知下游重新同步。这相当于在 GStreamer 层做了一次丢弃策略，避免了把太多帧塞给 OpenCV 内部处理导致压力过大。第二步，在代码层面彻底放弃cv2.VideoCapture(gst_pipeline)这种简单写法，改成直接用 GStreamer 的 Python binding (gi.repository.Gst) 创建管道，从appsink拉取GstSample，再手动转成 numpy 数组。虽然代码量多了不少，但整个过程完全可控，不会再有隐藏在 OpenCV C 代码里的那些不可预期行为。

import gi gi.require_version('Gst', '1.0') from gi.repository import Gst, GLib Gst.init(None) pipeline = Gst.parse_launch( "rtspsrc location=rtsp://user:pass@ip:554/Streaming/Channels/102 latency=0 ! " "rtph264depay ! h264parse ! v4l2h264dec ! videoconvert ! " "video/x-raw,format=RGB,width=640,height=640 ! " "queue max-size-buffers=2 leaky=downstream ! appsink name=sink" ) appsink = pipeline.get_by_name("sink") pipeline.set_state(Gst.State.PLAYING) def pull_frame(): sample = appsink.emit("try-pull-sample", 0) if not sample: return None buf = sample.get_buffer() caps = sample.get_caps() # 解析 width/height/stride，再 memoryview -> numpy 数组 ...

这段代码看起来头疼，但它确实稳定。从这个坑之后，我总结了一个原则：在嵌入平台上，如果某个库的封装层表现出不确定的崩溃，就直接绕开封装层，用底层接口自己管理资源。OpenCV 是好东西，但它对 GStreamer 的封装并不是所有版本都靠谱，你一定要在自己的板子上用长时间压力测试来验证，而不是在笔记本上跑一分钟没问题就认为万事大吉。

再说一个这个坑带出来的次生坑：颜色翻转。前期我用 GStreamer 管道时，videoconvert后面的 caps 如果没写format=RGB，那么appsink拿到的是底层解码器默认的格式，有可能是I420或者NV12，OpenCV 读到这种数据以为是三通道 BGR，就出现了彩色的红蓝通道互换、整幅画面像底片一样的效果。所以再次提醒：在 GStreamer 管道的末尾一定要明确指定输出格式，不要偷懒省略。

5. 常见问题全记录与性能参考

这一路踩坑，我顺手记录了不少常见问题的现象和解决方案，整理成一个速查表供你参考。这些问题不一定每一个你都会遇到，但一旦遇到，你可以直接照表排查，不用像我一样一个一个试过去。

关于性能数据，我实测的环境是：RK3588 板子，Linux 系统，模型是 YOLOv5s 转 RKNN 的 INT8 版本，输入分辨率 640x640，摄像头取 960x540 子码流。在只跑一路摄像头、服务层单请求的情况下，CPU 占用大约 40%-60%，NPU 推理单帧耗时大约 25-40ms，加上取流和解码，整体端到端延迟大约在 80-120ms。这个延迟对监控预警类应用完全够用。如果把预处理和后处理也做一些 C 扩展级的优化，理论上可以压到 60ms 左右。但坦白说，为这几十毫秒折腾代码，性价比不高——真正的瓶颈永远在取流链路上，不在推理本身。

更新的方向我也替你想好了：如果你想在这个链路上升级到 YOLOv8s，RKNN-toolkit2 的较新版本已经支持导出，流程和 YOLOv5s 基本一致，输入输出张量的处理逻辑差异也不大。另外模型轻量化方面，YOLOv5s 本身已经算小，如果还想再压，可以试一下用知识蒸馏后的变体或者把输入分辨率降到 320x320，检测速度能再翻一倍，代价是远距离小目标召回明显下降，是否值得要看你的实际场景。

最后再分享一个我实际操作中的心得：这类嵌入式视觉项目，最怕的不是模型精度不够，而是视频链路不稳定。而视频链路的问题，十有八九不是摄像头的问题，也不是网络的问题，是你取流和解码的代码在某些系统状态下踩到了隐藏的 bug。所以，在你把取流、解码、推理、服务层这些组件拼起来之前，一定先把它们拆开，各自独立跑满几个小时的稳定性测试，再逐步合并。合并一次，验证一次。我就是因为前期跳过了“取流和解码独立压测”这一步，才在后面花了整整一个周末的时间去跟那个段错误搏斗。这个教训，希望读到这里的你不需要重新经历一遍。