简介一套面向RK3588/RK3588S平台的YOLOv5s对象检测C落地方案源码源自RKNpu2项目通过线程池对RKNN模型推理做异步调度并使用ReLU替换原激活函数使NPU利用率明显提升实测可达142帧/秒适合需要在边缘设备上实现实时视觉检测的嵌入式开发者参考。整个压缩包共52个文件、约8.1MB以h/hpp头文件、cc源文件、so动态库、sh脚本以及rknn模型文件为主头文件给出rkYolov5s类与线程池接口so为RKNN运行时sh脚本覆盖编译与性能锁定内容紧凑且目录清晰。已有212人学习/浏览属于小而精的工程示例。工程内包含可运行的构建与调频脚本、RKNN模型、前后处理源码和说明文档拿到后即可在RK3588环境按常规流程编译运行。对二次开发而言可直接复用rkYolov5s封装类、前后处理逻辑与异步流水线设计也可参照其架构将其他RKNN模型接入为边缘计算场景下的高效推理提供了完整范本。1. RK3588上的YOLOv5s为什么需要异步框架142帧/秒怎么读先纠正一个容易误判的数字142帧/秒不是单路视频在RK3588上一口气跑YOLOv5s的纯NPU推理帧率而是多路视频流经过线程池异步化之后的总吞吐。单路YOLOv5s 640×640输入在RK3588 NPU上做一次推理大约要22~28ms也就是单路极限35~45fps把四路视频同时喂进同一个框架让NPU始终有活干、CPU不再空等总吞吐才能推到140fps这个量级。这个框架解决的核心问题是同步调用rknn_run时CPU和NPU互相等待导致的利用率断崖——推理期间CPU空转取流线程卡在推理里一帧慢整条流水线跟着慢。它用线程池把「取流→预处理→推理→后处理」拆成四个独立阶段适合四路以上摄像头并发检测、RK3588边缘盒子、以及所有想把NPU跑满但又不想把逻辑卷进RKNN API里的C项目。2. 线程池与任务队列先把“异步”落地2.1 为什么选线程池而不是每路视频一个线程直接在采集线程里调用rknn_run代码最简单但帧率一定上不去。原因在于rknn_run是阻塞调用NPU执行期间CPU线程只能干等而采集线程被占住之后下一帧的V4L2 buffer来不及回填驱动层很快就丢帧。给每路视频单独开一个线程做推理逻辑上没问题但四路就是四个线程同时去抢同一个NPU缺一个统一的排队机制最终效果是四路各自为政、总吞吐还不如同步单路。线程池的价值是把「提交推理」和「执行推理」解耦。采集线程只负责把帧数据拷贝进预先分配好的输入buffer然后往队列里丢一个任务就算完事真正调用rknn_inputs_set和rknn_run的是固定数量的worker线程。这样做的直接收益是采集线程的阻塞时间从20ms级降到微秒级四路取流互不影响NPU的排队由worker线程统一调度不会出现同一时刻四个线程同时怼rknn_run的情况。2.2 任务结构与阻塞队列选型任务队列是这个框架的核心数据结构我先给出任务定义这是整个异步化的基础struct AsyncTask { rknn_context ctx; // 该路视频对应的 RKNN 上下文 rknn_input inputs[1]; // 预分配的输入描述 rknn_output outputs[1]; // 输出描述推理完成后取结果 uint64_t frame_seq; // 帧序号用来判断延迟和丢帧 std::functionvoid(DetectResult) callback; // 推理完成后的回调做 NMS 和结果上报 // 输入 buffer 复用池的索引用完要归还 int pool_index -1; };这个结构体里最容易被忽略的是pool_index。YOLOv5s输入是640×640×3的RGB数据约1.2MB如果每帧都new一份在长时间运行时内存碎片会非常难看。我一般给每路视频分配2~3个输入buffer轮换复用任务提交时从复用池里取一个回调执行完再由框架归还。队列选型上推荐moodycamel::ConcurrentQueue或者自己用std::mutexstd::condition_variable包一个有界队列。后者更可控也便于排查问题class TaskQueue { public: explicit TaskQueue(size_t capacity) : capacity_(capacity) {} bool Push(AsyncTask task) { std::unique_lockstd::mutex lock(mutex_); not_full_.wait(lock, [this] { return queue_.size() capacity_ || stop_; }); if (stop_) return false; queue_.push(std::move(task)); not_empty_.notify_one(); return true; } bool Pop(AsyncTask task) { std::unique_lockstd::mutex lock(mutex_); not_empty_.wait(lock, [this] { return !queue_.empty() || stop_; }); if (stop_ queue_.empty()) return false; task std::move(queue_.front()); queue_.pop(); not_full_.notify_one(); return true; } void Shutdown() { std::unique_lockstd::mutex lock(mutex_); stop_ true; not_empty_.notify_all(); not_full_.notify_all(); } private: std::mutex mutex_; std::condition_variable not_empty_; std::condition_variable not_full_; std::queueAsyncTask queue_; size_t capacity_; bool stop_ false; };这里有两个参数是实际调出来的。队列深度capacity设置为16比较合理四路视频每路最多允许4帧在途小于8时NPU偶尔会饿着大于32时延迟会明显变大画面上出现「追不上实时」的黏滞感。选择条件变量而不是自旋锁是因为RK3588的CPU功耗在边缘场景很敏感自旋虽然能把延迟压低几十微秒但四路跑下来核心温度会高不少。2.3 线程池核心循环代码与参数线程池本身不做任何跟RKNN相关的事它只负责从队列里拿任务、执行、回调这是保持框架可复用的关键。worker线程的核心循环如下void WorkerLoop(int worker_id) { while (!stop_.load(std::memory_order_acquire)) { AsyncTask task; if (!queue_.Pop(task)) { break; } // 状态记录进队时间 vs 开始推理时间 task.begin_ts GetSteadyMicros(); int ret rknn_inputs_set(task.ctx, 1, task.inputs[0]); if (ret ! RKNN_SUCC) { LogError(rknn_inputs_set failed, ret%d, ret); RecycleInput(task.pool_index); continue; } ret rknn_run(task.ctx, nullptr); if (ret ! RKNN_SUCC) { LogError(rknn_run failed, ret%d, ret); RecycleInput(task.pool_index); continue; } ret rknn_outputs_get(task.ctx, 1, task.outputs[0], nullptr); if (ret ! RKNN_SUCC) { LogError(rknn_outputs_get failed, ret%d, ret); RecycleInput(task.pool_index); continue; } // 核心output 必须在拿到后立即交给回调不能拖到 release 之后 task.callback(std::move(task)); rknn_outputs_release(task.ctx, 1, task.outputs[0]); RecycleInput(task.pool_index); } }几个参数说清楚。worker线程数量建议4个也就是RK3588四个大核的数量。不是越多越好NPU只有一个worker多了只在rknn_inputs_set和memcpy阶段并行反而增加调度开销。rknn_run的第二个参数传nullptr表示默认阻塞执行这里没有用异步标志原因是NPU本身已经串行排队再搞异步反而需要额外的完成事件机制复杂度上不划算。代码里每次rknn_inputs_set都只传1个输入因为YOLOv5s是单输入模型。outputs的want_float建议保持0让NPU直接输出uint8的量化结果后处理时再反量化如果设成1NPU会额外做一次int8到float的转换四路并发时这部分的耗时会被放大。callback里只做两件事把output数据拷贝到结果缓冲然后触发NMS线程。千万不能在callback里做同步耗时操作否则worker线程被占住队列长度会瞬间涨满。3. RKNN推理后端的异步封装从rknn_init到rknn_outputs_get3.1 RKNN API调用流程与异步化边界RKNN的C API整体上是同步模型rknn_init初始化上下文rknn_query查询输入输出信息rknn_inputs_set喂数据rknn_run执行rknn_outputs_get取结果。异步化不是去改这些API的调用方式而是把「每个视频帧都走一遍完整API流程」变成「每个worker线程反复走rknn_inputs_set→rknn_run→rknn_outputs_get」。关键边界在于rknn_context在RK3588上可以多实例创建但每个实例的推理还是在同一个NPU上排队。所以每路视频创建独立context是合理的这样输入输出配置互不干扰但worker线程池必须全局共享统一往NPU提交任务。框架初始化时每个context需要查询并记录自己的输入输出属性struct RknnContext { rknn_context ctx 0; int input_count 0; int output_count 0; rknn_tensor_attr input_attrs[3]; rknn_tensor_attr output_attrs[3]; size_t input_total_size 0; bool Init(const std::string model_path) { int ret rknn_init(ctx, model_path.c_str(), 0, 0, NULL); if (ret 0) return false; rknn_input_output_num io_num; ret rknn_query(ctx, RKNN_QUERY_IN_OUT_NUM, io_num, sizeof(io_num)); if (ret 0) return false; input_count io_num.n_input; output_count io_num.n_output; for (int i 0; i input_count; i) { input_attrs[i].index i; ret rknn_query(ctx, RKNN_QUERY_INPUT_ATTR, input_attrs[i], sizeof(rknn_tensor_attr)); if (ret 0) return false; // YOLOv5s 的输入通常是 NHWC 布局这里累加整个 tensor 的字节数 input_total_size input_attrs[i].n_elems * input_attrs[i].type_size; } for (int i 0; i output_count; i) { output_attrs[i].index i; output_attrs[i].want_float 0; ret rknn_query(ctx, RKNN_QUERY_OUTPUT_ATTR, output_attrs[i], sizeof(rknn_tensor_attr)); if (ret 0) return false; } return true; } };初始化完成后输入buffer按照input_total_size统一分配一次后面不再单独malloc。这里有个细节rknn_tensor_attr里的type_sizeuint8是1float是4YOLOv5s的RKNN模型常见配置是输入uint8、输出uint8所以input_total_size就是1×3×640×640×1约1.2MB。3.2 推理器核心类参数与复用把RKNN封装成一个InferenceEngine类对外只暴露Submit接口。内部维护一批RknnContext、一个TaskQueue、一组worker线程、一个输入buffer复用池class RknnEngine { public: RknnEngine(size_t worker_count, size_t queue_capacity) : queue_(queue_capacity), stop_(false) { for (size_t i 0; i worker_count; i) { workers_.emplace_back([this] { WorkerLoop(); }); } } bool AddModel(const std::string model_path, int io_buffer_count 3) { RknnContext ctx; if (!ctx.Init(model_path)) return false; contexts_.push_back(std::move(ctx)); // 为当前模型分配 io buffer 复用池 size_t per_buffer contexts_.back().input_total_size; std::vectoruint8_t* bufs; for (int i 0; i io_buffer_count; i) { uint8_t* p new uint8_t[per_buffer]; bufs.push_back(p); free_buffers_.push(p); } io_pools_.push_back(std::move(bufs)); return true; } bool Submit(size_t model_index, const uint8_t* rgb_data, uint64_t seq, std::functionvoid(DetectResult) callback) { RknnContext ctx contexts_[model_index]; auto pool io_pools_[model_index]; if (free_buffers_.empty()) { // 池空说明 NPU 处理不过来直接丢帧比排队更合理 return false; } uint8_t* buf free_buffers_.front(); free_buffers_.pop(); memcpy(buf, rgb_data, ctx.input_total_size); AsyncTask task; task.ctx ctx.ctx; task.frame_seq seq; task.callback std::move(callback); task.inputs[0].index 0; task.inputs[0].type RKNN_TENSOR_UINT8; task.inputs[0].fmt RKNN_TENSOR_NHWC; task.inputs[0].size ctx.input_total_size; task.inputs[0].buf buf; bool ok queue_.Push(std::move(task)); if (!ok) { free_buffers_.push(buf); } return ok; } private: TaskQueue queue_; std::vectorstd::thread workers_; std::vectorRknnContext contexts_; std::vectorstd::vectoruint8_t* io_pools_; std::queueuint8_t* free_buffers_; std::atomicbool stop_; };注意fmt字段。这里写的是RKNN_TENSOR_NHWC因为在RK3588上YOLOv5s用NHWC布局喂数据比NCHW省一次转置。如果模型导出时指定的是NCHW这里就必须改成NCHW并配合reorder否则rknn_inputs_set会报参数错误。3.3 提交任务与回调把NPU变成“黑匣子”Submit接口的设计思想是调用方只提供RGB数据、帧号和回调完全不感知RKNN的存在。这样上层V4L2取流线程和后处理线程都可以独立演进换模型、换输入格式都不影响调用方代码。回调阶段的坑在于结果的所有权。rknn_outputs_get拿到的指针指向NPU内部buffer这个buffer在下次rknn_inputs_set时会被覆盖。所以callback的第一步必须是深拷贝输出数据而不是保存指针void PostProcessCallback(AsyncTask task) { DetectResult result; result.frame_seq task.frame_seq; // 深拷贝三个输出头的数据YOLOv5s 一般是 [1, 255, 80, 80]、[1, 255, 40, 40]、[1, 255, 20, 20] for (int i 0; i 3; i) { size_t bytes task.outputs[i].size; result.feature_maps[i].resize(bytes); memcpy(result.feature_maps[i].data(), task.outputs[i].buf, bytes); } // 丢给 NMS 线程不在这里做解码 g_nms_queue.Push(std::move(result)); }三个输出头的尺寸要和模型导出时对齐。如果用的是yolov5s.onnx转RKNN输出shape一般是[batch, 3, 80, 80, 85]这种带anchor的维度RKNN导出后会被压平成四维具体以rknn_query拿到的n_dims为准。调试阶段建议把output_attrs里的dims全部打印出来确认后再写死。4. V4L2多路取流与帧管理四路视频怎么喂满NPU4.1 V4L2采集线程的帧结构异步推理框架只解决「算」的问题帧从哪来是另一个关键环节。在RK3588上常见摄像头接口是MIPI CSI或USB驱动层都是V4L2。采集线程的职责是持续从驱动拿帧、转成RGB888、然后调用RknnEngine::Submit。这个线程必须做到不让NPU饿着但也不能让帧堆积。V4L2采集初始化有几处参数必须核对直接决定采集帧率上限int fd open(/dev/video0, O_RDWR); struct v4l2_format fmt; memset(fmt, 0, sizeof(fmt)); fmt.type V4L2_BUF_TYPE_VIDEO_CAPTURE; fmt.fmt.pix.width 1280; fmt.fmt.pix.height 720; fmt.fmt.pix.pixelformat V4L2_PIX_FMT_NV12; ioctl(fd, VIDIOC_S_FMT, fmt); struct v4l2_requestbuffers req; memset(req, 0, sizeof(req)); req.count 4; // 环形 buffer 数量 req.type V4L2_BUF_TYPE_VIDEO_CAPTURE; req.memory V4L2_MEMORY_MMAP; ioctl(fd, VIDIOC_REQBUFS, req); // mmap 映射驱动侧的 bufferoffset 是驱动返回的物理偏移 void* mmap_ptr[4]; for (int i 0; i 4; i) { struct v4l2_buffer buf; memset(buf, 0, sizeof(buf)); buf.type V4L2_BUF_TYPE_VIDEO_CAPTURE; buf.memory V4L2_MEMORY_MMAP; buf.index i; ioctl(fd, VIDIOC_QUERYBUF, buf); mmap_ptr[i] mmap(nullptr, buf.length, PROT_READ | PROT_WRITE, MAP_SHARED, fd, buf.m.offset); ioctl(fd, VIDIOC_QBUF, buf); // 全部入队 } enum v4l2_buf_type type V4L2_BUF_TYPE_VIDEO_CAPTURE; ioctl(fd, VIDIOC_STREAMON, type);req.count建议4这是驱动侧环形队列深度。太少会导致驱动无buffer可用而丢帧太多会增大采集延迟。注意CLOSED_CAPTION或者2路1080p同时采集时RK3588的VIPP通道数量有限需要确认用的是VICAP还是ISP通路。4.2 环形缓冲区与帧元数据取流循环里最容易写错的是DQBUF和QBUF的使用。DQBUF取出一个已经填满的buffer处理完必须马上QBUF还回去否则驱动可用buffer数减一几个周期后就会卡死while (running) { struct v4l2_buffer buf; memset(buf, 0, sizeof(buf)); buf.type V4L2_BUF_TYPE_VIDEO_CAPTURE; buf.memory V4L2_MEMORY_MMAP; int ret ioctl(fd, VIDIOC_DQBUF, buf); if (ret 0) { usleep(1000); continue; } uint8_t* nv12 static_castuint8_t*(mmap_ptr[buf.index]) buf.offset; // 转 RGB888这里走的是 NEON 优化版一帧 720p 约 4~6ms // 转出来的数据直接提交给推理引擎Submit 内部会 memcpy 到输入复用池 uint8_t* rgb swap_buffer_; NV12ToRGB888(nv12, rgb, 1280, 720); uint64_t seq frame_counter_; bool ok engine_.Submit(0, rgb, seq, callback); // 关键不管 Submit 成功与否QBUF 必须马上执行 ioctl(fd, VIDIOC_QBUF, buf); if (!ok) { Stats::Instance().AddDropped(seq); } }提交之后立刻QBUF推理引擎在拷贝时使用的是自己的输入复用池不依赖V4L2 buffer的存活时间。这是异步带来的优势换同步方案必须先等推理结束才能还buffer采集线程自然被卡死。帧元数据里要带上seq和采集时间戳用于延迟统计。回调阶段记录「采集时间→推理完成时间」的差值就能算出端到端延迟。这个指标比帧率更能反映流畅度至少需要维持在120ms以内否则开车看实时画面会有明显迟滞。4.3 绑核与CPU/内存带宽分配RK3588是44大小核架构大核是Cortex-A76小核是A55。四路1080p转RGB888、NMS、以及线程池worker全部挤在大核上时有可能出现CPU瓶颈。我一般把取流和图像转换线程绑在A76大核上线程池worker也绑大核NMS和统计线程绑A55避免它们抢占取流时间片// worker 线程绑到 4 个大核之一 cpu_set_t cpuset; CPU_ZERO(cpuset); CPU_SET(worker_id 4, cpuset); // RK3588 的 CPU4~CPU7 是大核 pthread_setaffinity_np(pthread_self(), sizeof(cpuset), cpuset);这里的worker_id 4是RK3588的CPU编号约定CPU0~3是小核CPU4~7是大核。如果跑的是rk3588s或rk3576核编号会不一样建议先读/proc/cpuinfo确认。绑核能稳定帧间隔但不能解决NPU排队本身的问题那是线程池队列的工作。内存带宽在四路720p以上场景会成为隐形的瓶颈。RK3588的内存控制器带宽有限四路NV12转RGB888同时进行时DMA和CPU搬运会争抢带宽。实测四路720p30时NV12ToRGB888的NEON优化版本大约占掉单核70%的时间如果转成浮点RGB再喂给NPU带宽占用直接翻倍。所以输入尽量保持uint8不要为了预处理方便转成float。5. 常见问题与排查五条避开就能稳定的坑5.1 开播瞬间延迟从30ms飙到120ms现象四路视频同时启动时前1~2秒画面延迟明显增大流畅度骤降之后恢复。原因所有采集线程同时提交第一批任务线程池队列被瞬间填满worker线程来不及消费NPU排队深度拉满延迟自然飙升。这是典型的「突发流量」问题不是算力不够。解决在Submit接口增加一个「每路在途帧数」的限流计数。每路视频最多允许2个任务在队列中超过直接丢帧保证新任务总是替换旧任务而不是排在后面// 每路维护一个 atomic 计数器task 进队 1回调执行完 -1 if (inflight_.fetch_add(1) 2) { inflight_.fetch_sub(1); return false; // 丢帧保持队列新鲜 }5.2 rknn_inputs_set返回参数错误但代码不崩现象程序能跑起来但rknn_inputs_set偶尔返回RKNN_ERR_PARAM_INVALIDperf里看不出明显异常。原因输入tensor的size和fmt与实际模型不匹配。最常见的是模型导出时输入是NCHW但代码里写死NHWC或者输入size写了整张图的字节数但RKNN期望的是单通道的size。解决在初始化阶段打印rknn_query拿到的实际attr逐字段核对rknn_inputs_set index0 fmtNHWC size1228800 实际attr: fmtNHWC, dims[1,640,640,3], typeUINT8有一个容易忽略的点如果size填错rknn_inputs_set不会立即报错可能会在rknn_run阶段返回错误排查时先怀疑inputs再怀疑run。5.3 memcpy和memset占用40% CPU现象htop里能看到某个worker线程CPU占用长期90%以上perf top显示memcpy占大头。原因每个任务Submit时做一次帧拷贝回调时深拷贝output又做一次NMS解码时还要拷贝。三层拷贝在四路并发下放大了数倍。解决从资源利用率的角度核心是减少无条件拷贝。输入侧用双缓冲加指针交换而不是每帧memcpy输出侧只在回调里拷贝且使用预分配的vector池避免频繁resize。实测把输入memcpy去掉之后单worker的CPU占用能下降15%左右。5.4 V4L2取流越跑越卡画面出现卡顿现象程序刚启动正常运行几分钟后帧率掉到20fps以下重启恢复。原因DQBUF和QBUF没有严格成对或者某个异常分支提前continue导致buffer没归还。驱动可用buffer数持续减少最终进入等待状态。解决检查所有DQBUF后的分支确认每个分支都会执行QBUF// 推荐用一个 RAII 包装器析构时自动归还 struct V4l2BufferGuard { int fd; v4l2_buffer buf; ~V4l2BufferGuard() { ioctl(fd, VIDIOC_QBUF, buf); } };5.5 四路并发时NPU利用率只有60%现象单路帧率正常四路加起来达不到预期总吞吐rknn_run的间隔有大段空隙。原因worker线程之间没有做任务粘性导致NPU排队和空闲交替出现。四个worker同时从队列里抢任务抢到后各自调用rknn_run但NPU一次只能执行一个其余三个都在排队等待排队期间NPU可能已经空闲。解决把worker和context做绑定每个worker固定处理一路上下文避免同一路任务被两个worker抢占。这样每路的任务序列天然串行化NPU的排队更平滑// 初始化时指定 worker 与 model 的对应关系 for (size_t i 0; i worker_count; i) { size_t model_index i % context_count; workers_.emplace_back([this, model_index] { WorkerLoop(model_index); }); }6. 验证与进阶调优从帧率倒推瓶颈框架装好之后第一步不是调参而是复现142帧/秒这个数字。我在工程里保留了一个极简的帧率统计模块不依赖任何外部工具class FpsCounter { public: void Tick(uint64_t seq, uint64_t submit_us, uint64_t done_us) { if (first_ts_ 0) first_ts_ submit_us; frames_; total_latency_us_ (done_us - submit_us); if (submit_us - last_report_ts_ 1000000) { double fps frames_ * 1000000.0 / (submit_us - first_ts_); double avg_latency total_latency_us_ / frames_; printf(fps%.2f, avg_latency%.2fms, seq%llu\n, fps, avg_latency / 1000.0, seq); frames_ 0; first_ts_ 0; total_latency_us_ 0; last_report_ts_ submit_us; } } };跑起来之后用htop观察四个worker线程的CPU占用和NPU排队通常出现两种瓶颈第一worker线程CPU跑满但NPU空闲。说明rknn_inputs_set或者memcpy占了太多时间优先查输入拷贝能去掉就去掉不能去掉就换NEON版本的memcpy。第二worker线程大量在等待NPU占用率也很高。说明任务队列深度太浅NPU会有短暂饥饿把queue_capacity从16调到24再观察延迟曲线。一个实用的调优顺序先固定worker4、queue16跑四路720p然后逐一增加队列深度记录总吞吐和延迟再测试worker6看是否有提升。绝大多数场景下worker4、队列深度在16~24之间是最稳的组合超过这个范围收益很小反而增加线程切换开销。从那以后我每次在RK3588上部署多路推理都强制走一遍「先确认数字构成→再压队列深度→最后调worker绑核」的验证流程而不是拿到代码就直接跑跑完发现慢得没法看又回头找框架的毛病。资源的价值在于给你一个正确的异步框架和那些不容易一次想明白的参数映射剩下的适配工作还是得靠你自己在板子上跑几轮、看几个perf报告。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/28 22:51:43 · 阅读 1.2k RK3588多路视频YOLOv5s异步推理框架：线程池与RKNN优化实践
简介一套面向RK3588/RK3588S平台的YOLOv5s对象检测C落地方案源码源自RKNpu2项目通过线程池对RKNN模型推理做异步调度并使用ReLU替换原激活函数使NPU利用率明显提升实测可达142帧/秒适合需要在边缘设备上实现实时视觉检测的嵌入式开发者参考。整个压缩包共52个文件、约8.1MB以h/hpp头文件、cc源文件、so动态库、sh脚本以及rknn模型文件为主头文件给出rkYolov5s类与线程池接口so为RKNN运行时sh脚本覆盖编译与性能锁定内容紧凑且目录清晰。已有212人学习/浏览属于小而精的工程示例。工程内包含可运行的构建与调频脚本、RKNN模型、前后处理源码和说明文档拿到后即可在RK3588环境按常规流程编译运行。对二次开发而言可直接复用rkYolov5s封装类、前后处理逻辑与异步流水线设计也可参照其架构将其他RKNN模型接入为边缘计算场景下的高效推理提供了完整范本。1. RK3588上的YOLOv5s为什么需要异步框架142帧/秒怎么读先纠正一个容易误判的数字142帧/秒不是单路视频在RK3588上一口气跑YOLOv5s的纯NPU推理帧率而是多路视频流经过线程池异步化之后的总吞吐。单路YOLOv5s 640×640输入在RK3588 NPU上做一次推理大约要22~28ms也就是单路极限35~45fps把四路视频同时喂进同一个框架让NPU始终有活干、CPU不再空等总吞吐才能推到140fps这个量级。这个框架解决的核心问题是同步调用rknn_run时CPU和NPU互相等待导致的利用率断崖——推理期间CPU空转取流线程卡在推理里一帧慢整条流水线跟着慢。它用线程池把「取流→预处理→推理→后处理」拆成四个独立阶段适合四路以上摄像头并发检测、RK3588边缘盒子、以及所有想把NPU跑满但又不想把逻辑卷进RKNN API里的C项目。2. 线程池与任务队列先把“异步”落地2.1 为什么选线程池而不是每路视频一个线程直接在采集线程里调用rknn_run代码最简单但帧率一定上不去。原因在于rknn_run是阻塞调用NPU执行期间CPU线程只能干等而采集线程被占住之后下一帧的V4L2 buffer来不及回填驱动层很快就丢帧。给每路视频单独开一个线程做推理逻辑上没问题但四路就是四个线程同时去抢同一个NPU缺一个统一的排队机制最终效果是四路各自为政、总吞吐还不如同步单路。线程池的价值是把「提交推理」和「执行推理」解耦。采集线程只负责把帧数据拷贝进预先分配好的输入buffer然后往队列里丢一个任务就算完事真正调用rknn_inputs_set和rknn_run的是固定数量的worker线程。这样做的直接收益是采集线程的阻塞时间从20ms级降到微秒级四路取流互不影响NPU的排队由worker线程统一调度不会出现同一时刻四个线程同时怼rknn_run的情况。2.2 任务结构与阻塞队列选型任务队列是这个框架的核心数据结构我先给出任务定义这是整个异步化的基础struct AsyncTask { rknn_context ctx; // 该路视频对应的 RKNN 上下文 rknn_input inputs[1]; // 预分配的输入描述 rknn_output outputs[1]; // 输出描述推理完成后取结果 uint64_t frame_seq; // 帧序号用来判断延迟和丢帧 std::functionvoid(DetectResult) callback; // 推理完成后的回调做 NMS 和结果上报 // 输入 buffer 复用池的索引用完要归还 int pool_index -1; };这个结构体里最容易被忽略的是pool_index。YOLOv5s输入是640×640×3的RGB数据约1.2MB如果每帧都new一份在长时间运行时内存碎片会非常难看。我一般给每路视频分配2~3个输入buffer轮换复用任务提交时从复用池里取一个回调执行完再由框架归还。队列选型上推荐moodycamel::ConcurrentQueue或者自己用std::mutexstd::condition_variable包一个有界队列。后者更可控也便于排查问题class TaskQueue { public: explicit TaskQueue(size_t capacity) : capacity_(capacity) {} bool Push(AsyncTask task) { std::unique_lockstd::mutex lock(mutex_); not_full_.wait(lock, [this] { return queue_.size() capacity_ || stop_; }); if (stop_) return false; queue_.push(std::move(task)); not_empty_.notify_one(); return true; } bool Pop(AsyncTask task) { std::unique_lockstd::mutex lock(mutex_); not_empty_.wait(lock, [this] { return !queue_.empty() || stop_; }); if (stop_ queue_.empty()) return false; task std::move(queue_.front()); queue_.pop(); not_full_.notify_one(); return true; } void Shutdown() { std::unique_lockstd::mutex lock(mutex_); stop_ true; not_empty_.notify_all(); not_full_.notify_all(); } private: std::mutex mutex_; std::condition_variable not_empty_; std::condition_variable not_full_; std::queueAsyncTask queue_; size_t capacity_; bool stop_ false; };这里有两个参数是实际调出来的。队列深度capacity设置为16比较合理四路视频每路最多允许4帧在途小于8时NPU偶尔会饿着大于32时延迟会明显变大画面上出现「追不上实时」的黏滞感。选择条件变量而不是自旋锁是因为RK3588的CPU功耗在边缘场景很敏感自旋虽然能把延迟压低几十微秒但四路跑下来核心温度会高不少。2.3 线程池核心循环代码与参数线程池本身不做任何跟RKNN相关的事它只负责从队列里拿任务、执行、回调这是保持框架可复用的关键。worker线程的核心循环如下void WorkerLoop(int worker_id) { while (!stop_.load(std::memory_order_acquire)) { AsyncTask task; if (!queue_.Pop(task)) { break; } // 状态记录进队时间 vs 开始推理时间 task.begin_ts GetSteadyMicros(); int ret rknn_inputs_set(task.ctx, 1, task.inputs[0]); if (ret ! RKNN_SUCC) { LogError(rknn_inputs_set failed, ret%d, ret); RecycleInput(task.pool_index); continue; } ret rknn_run(task.ctx, nullptr); if (ret ! RKNN_SUCC) { LogError(rknn_run failed, ret%d, ret); RecycleInput(task.pool_index); continue; } ret rknn_outputs_get(task.ctx, 1, task.outputs[0], nullptr); if (ret ! RKNN_SUCC) { LogError(rknn_outputs_get failed, ret%d, ret); RecycleInput(task.pool_index); continue; } // 核心output 必须在拿到后立即交给回调不能拖到 release 之后 task.callback(std::move(task)); rknn_outputs_release(task.ctx, 1, task.outputs[0]); RecycleInput(task.pool_index); } }几个参数说清楚。worker线程数量建议4个也就是RK3588四个大核的数量。不是越多越好NPU只有一个worker多了只在rknn_inputs_set和memcpy阶段并行反而增加调度开销。rknn_run的第二个参数传nullptr表示默认阻塞执行这里没有用异步标志原因是NPU本身已经串行排队再搞异步反而需要额外的完成事件机制复杂度上不划算。代码里每次rknn_inputs_set都只传1个输入因为YOLOv5s是单输入模型。outputs的want_float建议保持0让NPU直接输出uint8的量化结果后处理时再反量化如果设成1NPU会额外做一次int8到float的转换四路并发时这部分的耗时会被放大。callback里只做两件事把output数据拷贝到结果缓冲然后触发NMS线程。千万不能在callback里做同步耗时操作否则worker线程被占住队列长度会瞬间涨满。3. RKNN推理后端的异步封装从rknn_init到rknn_outputs_get3.1 RKNN API调用流程与异步化边界RKNN的C API整体上是同步模型rknn_init初始化上下文rknn_query查询输入输出信息rknn_inputs_set喂数据rknn_run执行rknn_outputs_get取结果。异步化不是去改这些API的调用方式而是把「每个视频帧都走一遍完整API流程」变成「每个worker线程反复走rknn_inputs_set→rknn_run→rknn_outputs_get」。关键边界在于rknn_context在RK3588上可以多实例创建但每个实例的推理还是在同一个NPU上排队。所以每路视频创建独立context是合理的这样输入输出配置互不干扰但worker线程池必须全局共享统一往NPU提交任务。框架初始化时每个context需要查询并记录自己的输入输出属性struct RknnContext { rknn_context ctx 0; int input_count 0; int output_count 0; rknn_tensor_attr input_attrs[3]; rknn_tensor_attr output_attrs[3]; size_t input_total_size 0; bool Init(const std::string model_path) { int ret rknn_init(ctx, model_path.c_str(), 0, 0, NULL); if (ret 0) return false; rknn_input_output_num io_num; ret rknn_query(ctx, RKNN_QUERY_IN_OUT_NUM, io_num, sizeof(io_num)); if (ret 0) return false; input_count io_num.n_input; output_count io_num.n_output; for (int i 0; i input_count; i) { input_attrs[i].index i; ret rknn_query(ctx, RKNN_QUERY_INPUT_ATTR, input_attrs[i], sizeof(rknn_tensor_attr)); if (ret 0) return false; // YOLOv5s 的输入通常是 NHWC 布局这里累加整个 tensor 的字节数 input_total_size input_attrs[i].n_elems * input_attrs[i].type_size; } for (int i 0; i output_count; i) { output_attrs[i].index i; output_attrs[i].want_float 0; ret rknn_query(ctx, RKNN_QUERY_OUTPUT_ATTR, output_attrs[i], sizeof(rknn_tensor_attr)); if (ret 0) return false; } return true; } };初始化完成后输入buffer按照input_total_size统一分配一次后面不再单独malloc。这里有个细节rknn_tensor_attr里的type_sizeuint8是1float是4YOLOv5s的RKNN模型常见配置是输入uint8、输出uint8所以input_total_size就是1×3×640×640×1约1.2MB。3.2 推理器核心类参数与复用把RKNN封装成一个InferenceEngine类对外只暴露Submit接口。内部维护一批RknnContext、一个TaskQueue、一组worker线程、一个输入buffer复用池class RknnEngine { public: RknnEngine(size_t worker_count, size_t queue_capacity) : queue_(queue_capacity), stop_(false) { for (size_t i 0; i worker_count; i) { workers_.emplace_back([this] { WorkerLoop(); }); } } bool AddModel(const std::string model_path, int io_buffer_count 3) { RknnContext ctx; if (!ctx.Init(model_path)) return false; contexts_.push_back(std::move(ctx)); // 为当前模型分配 io buffer 复用池 size_t per_buffer contexts_.back().input_total_size; std::vectoruint8_t* bufs; for (int i 0; i io_buffer_count; i) { uint8_t* p new uint8_t[per_buffer]; bufs.push_back(p); free_buffers_.push(p); } io_pools_.push_back(std::move(bufs)); return true; } bool Submit(size_t model_index, const uint8_t* rgb_data, uint64_t seq, std::functionvoid(DetectResult) callback) { RknnContext ctx contexts_[model_index]; auto pool io_pools_[model_index]; if (free_buffers_.empty()) { // 池空说明 NPU 处理不过来直接丢帧比排队更合理 return false; } uint8_t* buf free_buffers_.front(); free_buffers_.pop(); memcpy(buf, rgb_data, ctx.input_total_size); AsyncTask task; task.ctx ctx.ctx; task.frame_seq seq; task.callback std::move(callback); task.inputs[0].index 0; task.inputs[0].type RKNN_TENSOR_UINT8; task.inputs[0].fmt RKNN_TENSOR_NHWC; task.inputs[0].size ctx.input_total_size; task.inputs[0].buf buf; bool ok queue_.Push(std::move(task)); if (!ok) { free_buffers_.push(buf); } return ok; } private: TaskQueue queue_; std::vectorstd::thread workers_; std::vectorRknnContext contexts_; std::vectorstd::vectoruint8_t* io_pools_; std::queueuint8_t* free_buffers_; std::atomicbool stop_; };注意fmt字段。这里写的是RKNN_TENSOR_NHWC因为在RK3588上YOLOv5s用NHWC布局喂数据比NCHW省一次转置。如果模型导出时指定的是NCHW这里就必须改成NCHW并配合reorder否则rknn_inputs_set会报参数错误。3.3 提交任务与回调把NPU变成“黑匣子”Submit接口的设计思想是调用方只提供RGB数据、帧号和回调完全不感知RKNN的存在。这样上层V4L2取流线程和后处理线程都可以独立演进换模型、换输入格式都不影响调用方代码。回调阶段的坑在于结果的所有权。rknn_outputs_get拿到的指针指向NPU内部buffer这个buffer在下次rknn_inputs_set时会被覆盖。所以callback的第一步必须是深拷贝输出数据而不是保存指针void PostProcessCallback(AsyncTask task) { DetectResult result; result.frame_seq task.frame_seq; // 深拷贝三个输出头的数据YOLOv5s 一般是 [1, 255, 80, 80]、[1, 255, 40, 40]、[1, 255, 20, 20] for (int i 0; i 3; i) { size_t bytes task.outputs[i].size; result.feature_maps[i].resize(bytes); memcpy(result.feature_maps[i].data(), task.outputs[i].buf, bytes); } // 丢给 NMS 线程不在这里做解码 g_nms_queue.Push(std::move(result)); }三个输出头的尺寸要和模型导出时对齐。如果用的是yolov5s.onnx转RKNN输出shape一般是[batch, 3, 80, 80, 85]这种带anchor的维度RKNN导出后会被压平成四维具体以rknn_query拿到的n_dims为准。调试阶段建议把output_attrs里的dims全部打印出来确认后再写死。4. V4L2多路取流与帧管理四路视频怎么喂满NPU4.1 V4L2采集线程的帧结构异步推理框架只解决「算」的问题帧从哪来是另一个关键环节。在RK3588上常见摄像头接口是MIPI CSI或USB驱动层都是V4L2。采集线程的职责是持续从驱动拿帧、转成RGB888、然后调用RknnEngine::Submit。这个线程必须做到不让NPU饿着但也不能让帧堆积。V4L2采集初始化有几处参数必须核对直接决定采集帧率上限int fd open(/dev/video0, O_RDWR); struct v4l2_format fmt; memset(fmt, 0, sizeof(fmt)); fmt.type V4L2_BUF_TYPE_VIDEO_CAPTURE; fmt.fmt.pix.width 1280; fmt.fmt.pix.height 720; fmt.fmt.pix.pixelformat V4L2_PIX_FMT_NV12; ioctl(fd, VIDIOC_S_FMT, fmt); struct v4l2_requestbuffers req; memset(req, 0, sizeof(req)); req.count 4; // 环形 buffer 数量 req.type V4L2_BUF_TYPE_VIDEO_CAPTURE; req.memory V4L2_MEMORY_MMAP; ioctl(fd, VIDIOC_REQBUFS, req); // mmap 映射驱动侧的 bufferoffset 是驱动返回的物理偏移 void* mmap_ptr[4]; for (int i 0; i 4; i) { struct v4l2_buffer buf; memset(buf, 0, sizeof(buf)); buf.type V4L2_BUF_TYPE_VIDEO_CAPTURE; buf.memory V4L2_MEMORY_MMAP; buf.index i; ioctl(fd, VIDIOC_QUERYBUF, buf); mmap_ptr[i] mmap(nullptr, buf.length, PROT_READ | PROT_WRITE, MAP_SHARED, fd, buf.m.offset); ioctl(fd, VIDIOC_QBUF, buf); // 全部入队 } enum v4l2_buf_type type V4L2_BUF_TYPE_VIDEO_CAPTURE; ioctl(fd, VIDIOC_STREAMON, type);req.count建议4这是驱动侧环形队列深度。太少会导致驱动无buffer可用而丢帧太多会增大采集延迟。注意CLOSED_CAPTION或者2路1080p同时采集时RK3588的VIPP通道数量有限需要确认用的是VICAP还是ISP通路。4.2 环形缓冲区与帧元数据取流循环里最容易写错的是DQBUF和QBUF的使用。DQBUF取出一个已经填满的buffer处理完必须马上QBUF还回去否则驱动可用buffer数减一几个周期后就会卡死while (running) { struct v4l2_buffer buf; memset(buf, 0, sizeof(buf)); buf.type V4L2_BUF_TYPE_VIDEO_CAPTURE; buf.memory V4L2_MEMORY_MMAP; int ret ioctl(fd, VIDIOC_DQBUF, buf); if (ret 0) { usleep(1000); continue; } uint8_t* nv12 static_castuint8_t*(mmap_ptr[buf.index]) buf.offset; // 转 RGB888这里走的是 NEON 优化版一帧 720p 约 4~6ms // 转出来的数据直接提交给推理引擎Submit 内部会 memcpy 到输入复用池 uint8_t* rgb swap_buffer_; NV12ToRGB888(nv12, rgb, 1280, 720); uint64_t seq frame_counter_; bool ok engine_.Submit(0, rgb, seq, callback); // 关键不管 Submit 成功与否QBUF 必须马上执行 ioctl(fd, VIDIOC_QBUF, buf); if (!ok) { Stats::Instance().AddDropped(seq); } }提交之后立刻QBUF推理引擎在拷贝时使用的是自己的输入复用池不依赖V4L2 buffer的存活时间。这是异步带来的优势换同步方案必须先等推理结束才能还buffer采集线程自然被卡死。帧元数据里要带上seq和采集时间戳用于延迟统计。回调阶段记录「采集时间→推理完成时间」的差值就能算出端到端延迟。这个指标比帧率更能反映流畅度至少需要维持在120ms以内否则开车看实时画面会有明显迟滞。4.3 绑核与CPU/内存带宽分配RK3588是44大小核架构大核是Cortex-A76小核是A55。四路1080p转RGB888、NMS、以及线程池worker全部挤在大核上时有可能出现CPU瓶颈。我一般把取流和图像转换线程绑在A76大核上线程池worker也绑大核NMS和统计线程绑A55避免它们抢占取流时间片// worker 线程绑到 4 个大核之一 cpu_set_t cpuset; CPU_ZERO(cpuset); CPU_SET(worker_id 4, cpuset); // RK3588 的 CPU4~CPU7 是大核 pthread_setaffinity_np(pthread_self(), sizeof(cpuset), cpuset);这里的worker_id 4是RK3588的CPU编号约定CPU0~3是小核CPU4~7是大核。如果跑的是rk3588s或rk3576核编号会不一样建议先读/proc/cpuinfo确认。绑核能稳定帧间隔但不能解决NPU排队本身的问题那是线程池队列的工作。内存带宽在四路720p以上场景会成为隐形的瓶颈。RK3588的内存控制器带宽有限四路NV12转RGB888同时进行时DMA和CPU搬运会争抢带宽。实测四路720p30时NV12ToRGB888的NEON优化版本大约占掉单核70%的时间如果转成浮点RGB再喂给NPU带宽占用直接翻倍。所以输入尽量保持uint8不要为了预处理方便转成float。5. 常见问题与排查五条避开就能稳定的坑5.1 开播瞬间延迟从30ms飙到120ms现象四路视频同时启动时前1~2秒画面延迟明显增大流畅度骤降之后恢复。原因所有采集线程同时提交第一批任务线程池队列被瞬间填满worker线程来不及消费NPU排队深度拉满延迟自然飙升。这是典型的「突发流量」问题不是算力不够。解决在Submit接口增加一个「每路在途帧数」的限流计数。每路视频最多允许2个任务在队列中超过直接丢帧保证新任务总是替换旧任务而不是排在后面// 每路维护一个 atomic 计数器task 进队 1回调执行完 -1 if (inflight_.fetch_add(1) 2) { inflight_.fetch_sub(1); return false; // 丢帧保持队列新鲜 }5.2 rknn_inputs_set返回参数错误但代码不崩现象程序能跑起来但rknn_inputs_set偶尔返回RKNN_ERR_PARAM_INVALIDperf里看不出明显异常。原因输入tensor的size和fmt与实际模型不匹配。最常见的是模型导出时输入是NCHW但代码里写死NHWC或者输入size写了整张图的字节数但RKNN期望的是单通道的size。解决在初始化阶段打印rknn_query拿到的实际attr逐字段核对rknn_inputs_set index0 fmtNHWC size1228800 实际attr: fmtNHWC, dims[1,640,640,3], typeUINT8有一个容易忽略的点如果size填错rknn_inputs_set不会立即报错可能会在rknn_run阶段返回错误排查时先怀疑inputs再怀疑run。5.3 memcpy和memset占用40% CPU现象htop里能看到某个worker线程CPU占用长期90%以上perf top显示memcpy占大头。原因每个任务Submit时做一次帧拷贝回调时深拷贝output又做一次NMS解码时还要拷贝。三层拷贝在四路并发下放大了数倍。解决从资源利用率的角度核心是减少无条件拷贝。输入侧用双缓冲加指针交换而不是每帧memcpy输出侧只在回调里拷贝且使用预分配的vector池避免频繁resize。实测把输入memcpy去掉之后单worker的CPU占用能下降15%左右。5.4 V4L2取流越跑越卡画面出现卡顿现象程序刚启动正常运行几分钟后帧率掉到20fps以下重启恢复。原因DQBUF和QBUF没有严格成对或者某个异常分支提前continue导致buffer没归还。驱动可用buffer数持续减少最终进入等待状态。解决检查所有DQBUF后的分支确认每个分支都会执行QBUF// 推荐用一个 RAII 包装器析构时自动归还 struct V4l2BufferGuard { int fd; v4l2_buffer buf; ~V4l2BufferGuard() { ioctl(fd, VIDIOC_QBUF, buf); } };5.5 四路并发时NPU利用率只有60%现象单路帧率正常四路加起来达不到预期总吞吐rknn_run的间隔有大段空隙。原因worker线程之间没有做任务粘性导致NPU排队和空闲交替出现。四个worker同时从队列里抢任务抢到后各自调用rknn_run但NPU一次只能执行一个其余三个都在排队等待排队期间NPU可能已经空闲。解决把worker和context做绑定每个worker固定处理一路上下文避免同一路任务被两个worker抢占。这样每路的任务序列天然串行化NPU的排队更平滑// 初始化时指定 worker 与 model 的对应关系 for (size_t i 0; i worker_count; i) { size_t model_index i % context_count; workers_.emplace_back([this, model_index] { WorkerLoop(model_index); }); }6. 验证与进阶调优从帧率倒推瓶颈框架装好之后第一步不是调参而是复现142帧/秒这个数字。我在工程里保留了一个极简的帧率统计模块不依赖任何外部工具class FpsCounter { public: void Tick(uint64_t seq, uint64_t submit_us, uint64_t done_us) { if (first_ts_ 0) first_ts_ submit_us; frames_; total_latency_us_ (done_us - submit_us); if (submit_us - last_report_ts_ 1000000) { double fps frames_ * 1000000.0 / (submit_us - first_ts_); double avg_latency total_latency_us_ / frames_; printf(fps%.2f, avg_latency%.2fms, seq%llu\n, fps, avg_latency / 1000.0, seq); frames_ 0; first_ts_ 0; total_latency_us_ 0; last_report_ts_ submit_us; } } };跑起来之后用htop观察四个worker线程的CPU占用和NPU排队通常出现两种瓶颈第一worker线程CPU跑满但NPU空闲。说明rknn_inputs_set或者memcpy占了太多时间优先查输入拷贝能去掉就去掉不能去掉就换NEON版本的memcpy。第二worker线程大量在等待NPU占用率也很高。说明任务队列深度太浅NPU会有短暂饥饿把queue_capacity从16调到24再观察延迟曲线。一个实用的调优顺序先固定worker4、queue16跑四路720p然后逐一增加队列深度记录总吞吐和延迟再测试worker6看是否有提升。绝大多数场景下worker4、队列深度在16~24之间是最稳的组合超过这个范围收益很小反而增加线程切换开销。从那以后我每次在RK3588上部署多路推理都强制走一遍「先确认数字构成→再压队列深度→最后调worker绑核」的验证流程而不是拿到代码就直接跑跑完发现慢得没法看又回头找框架的毛病。资源的价值在于给你一个正确的异步框架和那些不容易一次想明白的参数映射剩下的适配工作还是得靠你自己在板子上跑几轮、看几个perf报告。希望帮到你。本文还有配套的精品资源点击获取