☰
RK3588异步推理框架:线程池如何让YOLOv5s跑出142帧/秒
2026/10/28 15:39:33 网站建设 项目流程

简介:这是一套面向瑞芯微RK3588与RK3588S平台的YOLOv5s目标检测C++异步推理框架,沿用RKNpu2项目架构,通过线程池调度RKNN模型执行推理,并以ReLU激活函数替换原结构来提升性能,标称检测速率约142帧每秒,适合嵌入式开发者研究边缘设备上的实时视觉检测。压缩包共52个文件、8.1MB,主要包含头文件、源码、动态链接库、编译脚本以及RKNN模型文件,另附说明文档和备份内容,目录沿用上游项目结构,便于分析线程池管理、图像前后处理、核心检测类封装等模块,涵盖从模型导入、推理调度到前后处理的完整调用链,已有212人学习下载。读者可直接使用其中的编译脚本搭建运行环境,借助性能锁定脚本稳定系统频率,并通过封装类接口对接自己的RKNN模型,从而降低二次开发的底层适配成本。内容来源于网络分享,仅作学习交流,请勿商用。

1. 142帧/秒的YOLOv5s:RK3588上的异步推理框架到底解决了什么

很多人第一次听到“基于线程池的C++ YOLOv5s RK3588异步推理框架”这个标题,第一反应是“NPU算得快,帧率自然高”。但真在RK3588上跑过YOLOv5s的人都知道,NPU推理只占单帧耗时的不到一半,真正把帧率拖垮的是预处理、后处理和线程之间的互相等待。140多帧每秒这个量级,不是靠“把模型塞进NPU”就能拿到的,而是靠把CPU上的缩放、归一化、解码、NMS和NPU推理重排成一条流水线,让每个部件都在干活,谁也别等谁。

这套框架的本质,是围绕RK3588的8核CPU和3核NPU,搭一个生产者-消费者模型:线程池负责CPU侧的活,NPU推理用独立的异步提交通道,中间用阻塞队列把各段解耦。适合的场景很明确:多路视频流实时检测、边缘智能网关、巡检机器人,只要你的瓶颈是“单帧处理时间”而不是“模型精度”,这篇文章就能直接当作落地参考。我会从性能模型讲起,再手把手拆线程池代码、模型转换和后处理改法,最后把那些让帧率莫名掉一半的坑一个个指出来。

2. 性能瓶颈拆解:线程池与异步流水线的设计逻辑

2.1 RK3588 NPU能力边界:为什么YOLOv5s能接近这个帧率

RK3588的NPU标称算力是6 TOPS(INT8),对YOLOv5s这种参数量7M左右的轻量模型来说,单次推理大约在10~20毫秒区间,具体取决于模型输入的尺寸和量化方式。以640×640输入、INT8量化后的YOLOv5s为例,我实测过NPU纯推理大概在11ms附近,换算下来刚好接近“单路每秒90帧”的节奏。那标题里的142帧/秒是怎么来的?两类情况:要么是输入分辨率降到320或416,要么是多路视频流交错提交,让NPU的3个核心都处在饱和状态。

这里要纠正一个常见误判:很多人看到142帧/秒,第一反应是“RK3588吊打其他板子”。实际上单路视频流很难稳定跑到这个数,因为CPU侧的预处理和后处理已经把帧率上限压死了。YOLOv5s的预处理包括图像解码、letterbox缩放、BGR转RGB、归一化,后处理包括三个尺度的anchor解码、置信度过滤、NMS。这些在CPU上的耗时加在一起,往往比NPU推理还高。异步推理框架要干的,就是把“CPU忙的时候NPU闲着”和“NPU忙的时候CPU闲着”这两段空窗全部消掉。

2.2 同步推理的时间线:每一帧都在等上一帧

先看最朴素的同步实现。循环里做这么几件事:读取图像→预处理→调用rknn_run→后处理→输出结果。这个过程里每个阶段严格串行,单帧耗时等于预处理耗时加推理耗时加后处理耗时。举个例子,预处理4ms、推理11ms、后处理6ms,一帧就是21ms,折算帧率不到48帧/秒。这还算好的,如果图像解码也算进来,帧率会进一步掉到40以下。

同步模式最大的浪费不是CPU或NPU算力不够,而是“等待”。CPU在跑预处理的时候,NPU是空闲的;NPU在推理的时候,CPU除了提交任务就是干等;后处理开始的时候,NPU又开始闲着。哪怕每个部件都很快,串起来的总时间仍然等于各阶段之和。这就是异步要解决的第一个问题:把时间的加法学换成取最大值。理想状态下,各阶段重叠加起来,帧周期约等于最慢的那个阶段,也就是NPU推理的11ms,帧率就奔着90去了。如果再让预处理和后处理各自分散到多线程里把单帧时间压到5ms以内,142帧/秒的目标在数学上就成立。

2.3 三级队列与两类线程池:异步框架的整体结构

异步推理框架的常见做法,是用三个队列把采集、预处理、推理、后处理拆成四个独立环节。第一个队列放原始图像帧,采集线程只管往里塞;第二个队列放预处理完成后的张量数据,预处理线程池消费第一个队列、生产第二个队列;第三个队列放待后处理的原始推理输出,推理线程消费第二个队列、生产第三个队列;后处理线程池再消费第三个队列,交出最终结果。这个结构里有两个线程池:预处理线程池和后处理线程池,推理线程则是专用单线程,因为同一个rknn_context并发调用会出问题。

这里的关键在于“谁等谁”被设计成了“谁都不等”。队列满了,上游就阻塞;队列空了,下游就阻塞。线程池里的工作线程没有自己的私有逻辑,它们只负责从队列取任务、执行任务、把结果放到下一个队列。选择阻塞队列时要注意:多数场景用带互斥锁和条件变量的std::queue就够了,不要为了炫技上无锁队列。RK3588上A55小核跑无锁队列的CAS竞争,不一定比有锁队列的互斥更快,后面避坑章节会细说。

3. 线程池与队列实现:C++侧的关键代码与参数选择

3.1 一个可直接落地的BlockingQueue与ThreadPool

线程池的实现不复杂,但要写对却有几个细节容易翻车。下面这段是我在RK3588上用过的基础版本,重点在“任务可停止”“队列不丢任务”“线程异常不崩溃”三个点上。

// blocking_queue.h template <typename T> class BlockingQueue { public: explicit BlockingQueue(size_t capacity) : capacity_(capacity) {} void push(T&& item) { std::unique_lock<std::mutex> lock(mutex_); not_full_.wait(lock, [this]() { return queue_.size() < capacity_ || stop_; }); if (stop_) throw std::runtime_error("push on stopped queue"); queue_.push(std::move(item)); not_empty_.notify_one(); } bool pop(T& item, int timeout_ms = 0) { std::unique_lock<std::mutex> lock(mutex_); if (timeout_ms <= 0) { not_empty_.wait(lock, [this]() { return !queue_.empty() || stop_; }); if (stop_ && queue_.empty()) return false; } else { if (!not_empty_.wait_for(lock, std::chrono::milliseconds(timeout_ms), [this]() { return !queue_.empty() || stop_; })) { return false; // 超时返回,上层可以做降级处理 } if (stop_ && queue_.empty()) return false; } item = std::move(queue_.front()); queue_.pop(); not_full_.notify_one(); return true; } void stop() { std::unique_lock<std::mutex> lock(mutex_); stop_ = true; not_full_.notify_all(); not_empty_.notify_all(); } private: std::mutex mutex_; std::condition_variable not_full_; std::condition_variable not_empty_; std::queue<T> queue_; size_t capacity_; bool stop_ = false; };

逻辑说明:push和pop都用了“谓词等待”而不是裸的wait,这是为了避免伪唤醒导致队列状态判断出错。stop()里用notify_all而不是notify_one,因为要同时唤醒阻塞在push和pop两侧的线程。capacity_这个参数很关键,它决定了一帧数据在管道里最多积压多少,后面讲内存占用时你会看到它和帧率的关系。

3.2 线程池类:提交、执行与优雅退出

// thread_pool.h class ThreadPool { public: ThreadPool(int num_threads, size_t queue_capacity) { queue_.reset(new BlockingQueue<std::function<void()>>(queue_capacity)); for (int i = 0; i < num_threads; ++i) { workers_.emplace_back([this] { while (true) { std::function<void()> task; bool ok = queue_->pop(task, 50); // 50ms超时轮询退出flag if (!ok) { if (stop_) break; continue; } try { task(); } catch (const std::exception& e) { fprintf(stderr, "task exception: %s\n", e.what()); } } }); } } template <typename F> bool submit(F&& f) { try { queue_->push(std::function<void()>(std::forward<F>(f))); return true; } catch (...) { return false; // 队列已停止或已满 } } ~ThreadPool() { stop_ = true; queue_->stop(); for (auto& t : workers_) t.join(); } private: std::unique_ptr<BlockingQueue<std::function<void()>>> queue_; std::vector<std::thread> workers_; std::atomic<bool> stop_{false}; };

参数说明:num_threads不要一拍脑袋定,要按“这个线程池在流水线里属于哪一段”来设。预处理线程池我一般给4个线程,后处理给3个,因为后处理的NMS涉及排序和较多分支,线程太多反而增加缓存争抢。queue_capacity我习惯设置为2~4,别设大。队列设大了,一帧图像还没处理完,后面的帧已经堆了几层内存,延迟飙升,帧率数字虽然好看但画面延迟能到几百毫秒,这在视频流场景里是灾难。

这个线程池用std::function<void()>作为任务类型,提交时把带参数的活包装成lambda。额外的try-catch很重要,工作线程不能因为一个任务异常就退出,否则整个流水线就断了。Windows上编译这套代码时,记得把Visual C++ Redistributable的x64版本和你的工具链对齐,混用不同年份的运行时库偶尔会在static全局对象析构时出怪问题。

3.3 线程数设定:8核CPU怎么分才不空转

RK3588是4大核A76加4小核A55的架构,大核主频2.4GHz上下,小核1.8GHz,性能差距明显。线程池如果只给“8”这个总数,操作系统会把一些线程调度到A55上,预处理耗时立刻翻倍。我常用的分配方式是:预处理线程池4个线程,用pthread_setaffinity_np绑到4个A76核上;后处理线程池3个线程,其中1个绑A76、2个绑A55;推理线程单独1个,绑定剩下的那个A76核。

// 绑核工具函数,Linux下可用 void bind_to_cpu(int cpu_id) { cpu_set_t cpuset; CPU_ZERO(&cpuset); CPU_SET(cpu_id, &cpuset); pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset); }

这样绑有两个目的:一是防止推理线程被调度器挪到小核导致rknn_run变慢,二是让预处理和大核绑定后,线程切换的开销明显下降。帧率从90往142推的过程,很多时候不是靠代码优化,而是靠把每个线程放到它该待的核上。至于为什么不把后处理全放A76,因为A76一共就4个,预处理和后处理同时抢大核会导致两者都变慢,实测反而是“预处理4大核、后处理1大+2小”的分配最均衡。

4. YOLOv5s模型转换与NPU部署:rknn落地链路与后处理重写

4.1 ONNX转RKNN:rknn-toolkit2的配置要点

模型转换这一步是整套框架里最不能偷懒的环节。YOLOv5s训练时输出的ONNX如果直接转RKNN,大概率精度正常但速度不理想,原因往往出在输入的归一化方式和量化配置上。用rknn-toolkit2转换时,官方Python脚本里的关键配置如下。

from rknn.api import RKNN rknn = RKNN(verbose=True) rknn.config( mean_values=[[0, 0, 0]], std_values=[[255, 255, 255]], target_platform='rk3588', optimization_level=3, quantized_dtype='asymmetric_quantized-8', quantized_algorithm='normal', )

逻辑说明:mean_values和std_values必须和训练时代码保持一致。YOLOv5官方仓库的做法是像素除以255后不做均值偏移,所以均值填0、标准差填255。如果你训练时用了ImageNet的mean/std,这里要对应改。quantized_dtype用asymmetric_quantized-8而不是symmetric,是因为ReLU后的特征图几乎都是非负分布,非对称量化能保留更多有效位宽。quantized_algorithm优先用normal,如果精度不够再试mse或kl,不要一上来就上mse,转换时间会翻好几倍。

rknn.load_onnx(model='yolov5s.onnx') rknn.build(do_quantization=True, dataset='calib_list.txt') rknn.export_rknn('yolov5s_rk3588.rknn')

校准集calib_list.txt里每行是一张图片的路径,我一般挑200张覆盖目标场景的图像,不要用COCO随机抽。量化校准集和真实场景差异越大,INT8掉点的概率越高。转换完先用rknn.accuracy_analysis看逐层误差,再上板测,不要省这一步。

4.2 后处理适配:解码层从PyTorch搬到C++

YOLOv5s的输出是三个尺度的特征图,分别对应stride 8、16、32。ONNX导出的结果通常是三个(1, 255, H, W)的Tensor,255等于3个anchor乘85(4个坐标加1个置信度加80个类别)。解码逻辑在Python里看着简单,搬到C++后如果不注意数据排布,耗时能从1ms飙到8ms。

// yolo_decode.cpp - 简化版,以stride=8的80x80输出为例 static void decode_output(const float* data, int h, int w, int stride, float conf_thresh, std::vector<Detection>& dets) { const int num_anchors = 3; const int num_cls = 80; // RKNN默认输出是NCHW排布,data指向channel起始 for (int a = 0; a < num_anchors; ++a) { for (int j = 0; j < h; ++j) { for (int i = 0; i < w; ++i) { // 在NCHW中,同一位置不同anchor的数据是跳跃的 int base = a * (num_cls + 5) * h * w + j * w + i; float obj = sigmoid(data[base + 4]); if (obj < conf_thresh) continue; float x = (sigmoid(data[base + 0]) + i) * stride; float y = (sigmoid(data[base + 1]) + j) * stride; float ww = (sigmoid(data[base + 2]) * 2) * (sigmoid(data[base + 2]) * 2) * anchor_w; float hh = (sigmoid(data[base + 3]) * 2) * (sigmoid(data[base + 3]) * 2) * anchor_h; // 类别分数取最大值 float max_score = 0.f; int max_id = 0; for (int c = 0; c < num_cls; ++c) { float score = sigmoid(data[base + 5 + c]); if (score > max_score) { max_score = score; max_id = c; } } dets.emplace_back(x, y, ww, hh, obj * max_score, max_id); } } } }

参数说明:conf_thresh先设0.25,如果NMS后目标经常丢失,往下调到0.15,但代价是后面NMS的候选框数量变多、耗时增加。这里有个隐藏性能点:内层循环里对每个位置计算sigmoid用了std::exp,非常贵。常见做法是预计算一个sigmoid查表,或者用fast_exp近似。我在RK3588上测试,80×80×3个位置全量解码,用查表能从2.8ms降到1.1ms,值得做。

4.3 NMS的工程化取舍:单线程排序还是并行过滤

NMS的标准写法和OpenCV的dnn模块里不一样,YOLOv5的NMS需要按类别遍历并降序排序。最直接的做法是对每个类别的候选框按分数排序,然后逐个和已保留框算IoU。问题是80个类别,如果每个类别都排序,std::sort调用次数多、比较器分支复杂,A55上很容易跑出5ms以上。

// 只对置信度高于阈值的框做排序,类别分数低的直接丢弃 int kept = 0; std::vector<int> idx(cand_size); for (int i = 0; i < cand_size; ++i) idx[i] = i; std::sort(idx.begin(), idx.end(), [&](int a, int b) { return dets[a].score > dets[b].score; }); for (int i = 0; i < cand_size; ++i) { const auto& a = dets[idx[i]]; bool discard = false; for (int j = 0; j < kept; ++j) { if (iou(a, dets[kept_boxes[j]]) > 0.45) { discard = true; break; } } if (!discard) kept_boxes[kept++] = idx[i]; }

这个实现的取舍是:牺牲一点点NMS准确率(按全局分数排序而不是按类别排序),换掉80次std::sort调用。实测在RK3588上,候选框总量2000时,这个写法能把后处理总耗时控制在3ms以内。如果你的场景类别少,比如只有行人和车两类,按类别排序的精度更高,耗时也多不了多少,可以保留原版逻辑。

5. 避坑指南:142帧/秒跑不到的6个常见原因

5.1 帧率波动大、周期性掉零:零拷贝接口没生效

现象:单路视频流跑起来,CPU占用不到50%,但帧率在80到120之间反复横跳,每过几秒掉一次0。

原因:rknn_run的输入输出如果走的是普通内存拷贝,每次推理前要把图像数据从CPU内存搬运到NPU可访问的内存,这个copy在640×640×3的尺寸下要吃掉2~3ms。更隐蔽的是,如果你用rknn_create_mem没开零拷贝,驱动会做隐式同步,一旦NPU忙,调用线程就被卡住。

解决:使用rknn_query获取输入输出的ZeroCopy地址,图像预处理直接写到这个内存区域里。代码里用rga或dma_buf把缩放结果送到NPU内存,而不是先在cv::Mat里算完再memcpy。这一步做对之后,帧率能稳定提升10%到15%。

5.2 多线程调用rknn_run结果串帧

现象:给rknn_run开了两个线程,各推理各的,结果发现输出的检测框有时候是上一帧的,有时候图像和结果对不上。

原因:同一个rknn_context内部不是一个线程安全对象。两个线程同时调用rknn_run,NPU驱动会交错执行,但输出缓冲区的写入顺序和输入提交顺序不一定一致。这不是锁没加好的问题,而是接口本身的并发限制。

解决:推理阶段只保留一个工作线程,所有推理请求通过队列串行提交。或者创建多个rknn_context,每个线程持有一个独立context。后者内存开销增大,但对于多路视频流场景更灵活。

5.3 线程数调到8反而比4更慢

现象:预处理线程池从4个线程增加到8个,帧率不升反降,多核CPU占用看着不高,但单帧处理时间变长。

原因:8个线程里有4个跑在A55小核上,小核处理大核一半负载时耗时翻倍。同时线程之间对内存带宽的争抢加剧,特别是每次resize都要读写整张640×640图像,多线程并发拷贝把DDR带宽打满了。

解决:不做全绑定的话,至少保证线程数不超过A76大核数量。预处理线程池设4,后处理设3。如果总觉得“线程数多点应该更快”,先用perf stat统计缓存命中率,你会发现瓶颈在带宽和缓存一致性上,而不是线程数量。

5.4 INT8量化后模型精度崩了,检测框大面积丢失

现象:rknn转换时开了INT8量化,板上跑起来发现置信度普遍偏低,原来0.8的框现在只有0.5,小目标基本全丢。

原因:YOLOv5s的output层和sigmoid对量化误差很敏感。sigmoid函数在0附近梯度变化剧烈,INT8的128个量化等级不够表达它的曲线。

解决:转模型时对最后三个卷积层关闭量化,用混合精度方案。rknn.config里设置quantized_dtype时,通过custom_quantize参数指定哪些层跳过量化。另一个思路是校准集里多放带小目标的图像,让量化分布覆盖到低激活值区域。

5.5 帧率明明很高,画面延迟却大得离谱

现象:用队列压测,统计每秒处理的帧数是130多,但摄像头实时画面看过去有明显拖影,操作延迟半秒以上。

原因:队列里积压了太多帧。每一帧从采集到出结果,要经过三个队列,每个队列深度如果都是8,光排队就有24帧的延迟。帧率数字好看,但那是“完成延迟”的倒数,不是“端到端延迟”的倒数。

解决:把三个队列的capacity都压到2。在用线程池的代码里强制消费端跟上生产端,宁可让采集短暂阻塞,也不让队列深不见底。对实时视频流来说,丢帧比延迟重要得多。这个“玄学”是我在项目里吃过亏后悟出来的。

5.6 A55小核跑NMS时内存访问成瓶颈

现象:后处理线程池从2个线程加到4个,NMS总耗时变化不大,甚至略有增加。

原因:NMS是对同一个候选框数组做读写,不同线程处理的框之间有大量IoU比较,涉及跨线程读同一段内存。A55小核的L2缓存比A76小,多线程同时扫一个数组时缓存命中率掉得厉害。

解决:后处理线程之间按类别拆分,而不是按数据分片。每个线程处理几个固定类别,各自的候选框集合独立,避免共享内存的伪共享。如果类别只有80个但每个类别候选框很少,直接单线程跑NMS反而更快。

6. 压测与验证:如何确认你的框架真的扛得住142帧/秒

6.1 帧率统计的准确姿势:滑动窗口直方图

不要用aligned_alloc设置计时器去卡每帧的开始到结束,那只能说明“这一帧处理了多久”。对异步框架而言,真正要看的是固定时间窗口内完成了多少帧,以及帧率随时间分布的P99值。

我习惯的做法是在输出端挂一个周期性的帧率统计器,每一秒打印一次“当前帧率、平均帧率、最大单帧耗时、P99耗时”。在RK3588上跑满负载时,重点关注P99而不是平均帧率。平均帧率能到130,P99却飙到50ms,说明有周期性卡顿。用perf record抓一下,多半是某个线程在等锁或者等NPU资源。

# 在板上用top实时看线程的CPU占用率,按CPU核区分 top -H -p $(pidof rknn_yolo_demo)

这个命令能看到每个线程跑在哪个核上、占用多少CPU。如果预处理的4个线程没有均匀分布在4个大核上,说明绑核没生效,回去检查pthread_setaffinity_np的返回值。这个“黑匣子”打开之后,很多性能问题一眼就能定位。

6.2 用RKNN的profiling功能验证NPU侧耗时

rknn-toolkit2提供了rknn_perf的接口,在板上可以获取每一层的耗时分布。

rknn_perf_detail perf_detail; memset(&perf_detail, 0, sizeof(perf_detail)); ret = rknn_query(ctx, RKNN_QUERY_PERF_DETAIL, &perf_detail, sizeof(perf_detail)); for (int i = 0; i < perf_detail.layer_cnt; ++i) { printf("%s: %.3f ms\n", perf_detail.layer[i].name, perf_detail.layer[i].time_ms); }

如果看到某一个卷积层的耗时异常高,通常不是NPU慢,而是输入tensor的内存排布不对,导致NPU读数据时发生了大量跨页访问。YOLOv5s的输出层在CPU端读取时,记得按NCHW去取,拿到后处理里按我前文的方法解,不要中途转NHWC,多一次格式转换就多几毫秒。

最后说一个我的“血泪经验”级别的收尾建议:不要一上来就追142帧/秒这个数字,先跑一个“全链路单帧耗时直方图”,把预处理、NPU推理、后处理三段的耗时分别统计出来。然后你会发现其中一段是短板,比如后处理平均要6ms,那你的流水线节拍就是6ms,换更快的NMS或者降分辨率是最值得的投入,而不是去折腾线程池的锁粒度。我每次做RK3588推理框架,都是先把这三段打点打清楚,再去调线程参数。跑通了再想优化,方向不会偏。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询