☰
RK3588双路视觉实战:每路一个线程池的线程模型设计与调优
2026/10/31 17:24:59 网站建设 项目流程

这个系列写到第14篇,单路的yolov5s部署已经跑通了,RK3588上的模型转换、推理封装、显示链路也都搭好了。今天开始真正切双路视觉方案,标题写得很直白:两路各一个线程池。先别急着看代码,我想先聊聊为什么双路视觉这么容易写崩——很多人在这一步卡住,不是模型问题,也不是硬件不够,而是线程模型没设计好。把两路摄像头各开一个线程去跑,看起来简单,跑起来就是掉帧、卡死、内存暴涨三件套。这篇我把为什么这样设计、线程池参数怎么定、代码怎么落地、实测踩了哪些坑,一次说清楚。

1. 双路视觉的准入门槛:为什么两路摄像头这么难调

1.1 双路视觉到底在解决什么问题

双路视觉不是“多加一个摄像头”这么简单。我做这套方案的初衷是给机器人做前向双目感知,一路朝左前,一路朝右前,两路同时做目标检测;后来也有做工业质检的朋友问过,产线两侧各来一个摄像头,检测产品两面缺陷。不管是双目避障、AB面质检,还是双向来料检测,需求本质都一样:两个画面几乎同时进入系统,各自独立完成yolov5s推理,然后把结果汇总到主控。

单路跑通的时候,很多人会下意识觉得双路就是把代码复制一份,开两个线程,各跑各的。我在RK3588上试过这种最朴素的写法:两个std::thread,每个线程里都是“取帧→推理→画框→显示”的完整循环。结果跑起来之后,画面一卡一卡的,两路的帧率都不稳定,偶尔还会出现某一路完全黑屏、另一路正常的现象。

问题出在哪?因为两路之间不是真的“各跑各的”。它们共享同一颗CPU、同一个NPU、同一块内存带宽。摄像头的取流、图像的缩放归一化、NPU推理、后处理NMS,这些环节全都挤在一起。单路时资源够用看不出来,双路一加上,互相抢资源,谁都没跑好。

1.2 裸线程方案为什么扛不住双路推理

把单路代码直接复制成双线程,至少要面对三个问题。

第一,NPU是串行瓶颈。yolov5s在RK3588上走的是NPU推理,rknn_run这个调用是阻塞式的。线程A调用rknn_run的时候,线程B如果也调,B就得排队等NPU释放。结果两个线程从“并行”变成了“轮流用加速器”,整体吞吐和单路几乎一样,但每路的延迟还增加了,因为要等对方的推理结束。

第二,帧缓冲缺失。摄像头是以固定帧率往系统里推数据的,但推理速度不一定跟得上帧率。单线程方案里,线程被摄像头read和NPU推理两头阻塞,如果某一帧推理慢了,下一帧只能干等,帧率抖动直接被感知到。两路同时抖动,观感就是“两路都在抽搐”。

第三,线程生命周期管理。裸线程的创建销毁虽然开销不大,但双路方案一旦要动态启停某一路(比如临时关闭一路摄像头省电),代码就很容易写成一团乱麻。线程池把“任务”和“线程的生死”解耦,启停一路只需要往池子里提交或停止提交任务,不需要去操作一个裸线程对象。

所以线程池在这里不是炫技,而是双路方案里最自然的解耦手段。每路一个线程池,路的内部是“生产者往队列放帧,池里的线程取帧推理”,路与路之间互相隔离。想通这一点,双路视觉才真正迈过第一道门槛。

2. 线程池参数背后的硬件账本:先搞懂RK3588再调线程

2.1 RK3588的CPU与NPU结构决定了线程池边界

线程池不是随便设参数的,先要把RK3588的“家底”算清楚。

RK3588的CPU是8核架构:4个Cortex-A76大核 + 4个Cortex-A55小核。A76是性能核,适合做重计算;A55是效率核,功耗低但算力弱。如果你不管线程跑在哪个核上,调度器可能把两个采集线程丢到同一个大核,把两个推理线程也挤在一起,等于8核只用了3核,其余全闲着。

NPU这边,RK3588标称6 TOPS,是三核设计。yolov5s如果转成int8模型,单核跑一路绰绰有余,但双路同时推理时,三核NPU可以被两路任务分掉,也可以一路独占一个核、另一路用两个核。这里就有一个重要结论:NPU本身是能并行处理多路任务的,前提是你的rknn调用能被拆成多个task,并且线程池里同时提交了多个推理任务。如果只是两个线程轮流调同一个rknn context,NPU多核的优势也用不出来。

内存带宽也需要计入成本。双路的MIPI CSI或USB摄像头取流,加上两路图像的预处理,每一帧RGBA或RGB888的数据量都不小。如果摄像头是4K分辨率,一路就要吃掉不少内存带宽,两路叠加后CPU缓存命中率下降,线程调度延迟也会上升。这也是为什么线程池的线程数不能简单按“CPU核数+1”来设。

2.2 核心线程数、队列容量与拒绝策略的推荐值

线程池关键参数就三个:核心线程数、最大线程数、任务队列。

以我的实践为准,每路的线程池我建议这样配:

参数推荐值说明
核心线程数2~3一路至少有2个推理线程并行,yolov5s预处理和后处理可以流水
最大线程数3不要超过4,RK3588大核有限,超过之后上下文切换开销大于收益
任务队列类型有界队列容量8~16帧,防止帧堆积吃光内存
队列满时的策略丢弃最旧帧视觉任务宁可处理新帧,不处理积压旧帧

核心线程数给2~3,是因为一路的yolov5s推理可以拆成三段:预处理(letterbox缩放、归一化)、NPU推理、后处理(NMS)。两个线程可以一个做预处理的同时另一个做NPU推理或后处理,形成软件流水线。第三个线程可作为临时弹性,用于处理突发帧。

任务队列为什么选有界的?无界队列在帧率短暂波动时看似“吸收”了积压,但实际上是把内存风险延后了。摄像头的帧率波动是常态,推理偶尔卡顿也是常态。如果队列无限增长,帧在队列里等了几百毫秒才被处理,这个帧已经过时了。视觉检测要的是新鲜度,不是完整性。队列容量8~16帧,大约对应0.3到0.5秒的缓冲,足够抵抗瞬时波动,又不会积压太多旧帧。

队列满时选“丢弃最旧帧”而不是“丢弃新帧”,是从实测中得出的经验。摄像头30fps时,每帧间隔约33ms,队列积压说明推理跟不上采集。这时候保留最新帧,把最旧的丢掉,检测结果始终反映最近发生的情况;反过来丢弃新帧,画面会越来越迟钝,表现为延迟持续增加。

2.3 两路各建一个线程池,而不是共用一个

这是本篇标题的核心:两路各一个线程池。为什么不是把两路任务扔进同一个线程池?我一开始也是共用思路,感觉能省线程,跑起来问题不少。

共用线程池的第一个问题是任务饥饿。路0的推理任务如果一直占满线程,路1的任务就会排队等待。摄像头帧率不可能完全一致,某一路偶尔多出几帧,就会把另一路挤到队尾,结果一路流畅一路卡。两个独立线程池,路1的任务再忙也只占用路1的线程,路0不受影响。

第二个问题是故障隔离。共用池子里,如果一路的推理任务内部出现异常(比如rknn context失效、某帧数据处理超时),整个池子的线程可能被拖住,另一路也跟着遭殃。分开后,一路崩了最多停一路,另一路还能继续工作。

第三个好处是调参灵活。两路摄像头如果型号不同、分辨率不同,推理负载是不一样的。比如路0是1080p,路1是4K,路1的预处理耗时明显更高,那么路1的线程池可以配置更大的队列或更多的最大线程数,而路0保持轻量。共用一个池子没法这样单独调。

代价当然也有:两个池子的线程总数会多一些,CPU上下文切换更频繁。但把每路最大线程数控制在3以内,总线程数不超过6,再加上采集线程和显示线程,RK3588的8核是扛得住的。

3. 双路线程池落地代码:从帧采集到yolov5s推理

3.1 数据流水线:采集、缓冲、推理、输出

双路方案的整体数据流,我把每路设计成一条独立的流水线:

摄像头取流线程(生产者)→ 有界帧队列 → 推理线程池(消费者)→ 结果队列 → 主控显示/上报

每路有两个队列:一个是原始帧队列,一个是推理结果队列。采集线程只负责从摄像头拿帧、往帧队列放;线程池里的线程只负责从帧队列取帧、做推理、把结果放到结果队列;主线程在固定时间间隔内从两个结果队列取数据,用于显示或上报。

这样设计的好处是每个环节的单点阻塞不会传导到整条链。采集线程如果等待摄像头的read阻塞,推理线程池还能继续处理队列里的旧帧;线程池如果因为NPU忙而暂时跑不动,采集线程还能继续往队列放帧,只要队列有界,最坏情况就是丢弃旧帧。

具体到代码结构,我不建议用复杂的框架,C++11的std::thread + std::mutex + std::condition_variable就足够了。下面给出我实际使用的ThreadPool实现,这是常规标准写法,配合注释说明关键点。

3.2 一个可直接用的ThreadPool实现

这个ThreadPool类不算多复杂,但有几个细节容易写错,我标注一下。

// thread_pool.h #pragma once #include <atomic> #include <condition_variable> #include <deque> #include <functional> #include <future> #include <mutex> #include <thread> #include <vector> class ThreadPool { public: explicit ThreadPool(size_t threads) : stop_(false) { for (size_t i = 0; i < threads; ++i) { workers_.emplace_back([this] { while (true) { std::function<void()> task; { std::unique_lock<std::mutex> lock(queue_mutex_); cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); }); if (stop_ && tasks_.empty()) { return; } task = std::move(tasks_.front()); tasks_.pop_front(); } task(); } }); } } // 注意:这里用了deque,存取都在队尾,配合“队列满丢弃最旧帧”策略时, // 需要改为在队头pop并push到队尾,或者直接pop_front + push_back。 template <class F> auto enqueue(F&& f) -> std::future<decltype(f())> { using return_type = decltype(f()); auto task = std::make_shared<std::packaged_task<return_type()>>( std::forward<F>(f) ); std::future<return_type> res = task->get_future(); { std::unique_lock<std::mutex> lock(queue_mutex_); if (stop_) { throw std::runtime_error("enqueue on stopped ThreadPool"); } tasks_.emplace_back([task]() { (*task)(); }); } cv_.notify_one(); return res; } size_t pendingCount() { std::unique_lock<std::mutex> lock(queue_mutex_); return tasks_.size(); } ~ThreadPool() { { std::unique_lock<std::mutex> lock(queue_mutex_); stop_ = true; } cv_.notify_all(); for (std::thread& worker : workers_) { worker.join(); } } private: std::vector<std::thread> workers_; std::deque<std::function<void()>> tasks_; std::mutex queue_mutex_; std::condition_variable cv_; std::atomic<bool> stop_; };

这个实现的几个关键点:

  • 任务队列用std::deque而不是std::queue,因为后面实现“丢弃最旧帧”需要从队头操作。
  • wait的lambda条件是stop_或tasks_非空,这样shutdown时线程能正常退出,不会死等。
  • packaged_task用来拿返回值,便于上层知道推理任务是否完成,或者拿到推理结果指针。
  • 析构时先置stop_再notify_all,避免线程在等待中永远阻塞。

3.3 把yolov5s的预处理、推理、后处理打包成任务

双路方案里,提交给线程池的不是一个“普通的函数”,而是一个完整的推理任务。我把推理过程封装成一个lambda,捕获帧数据和推理上下文指针。

using FramePtr = std::shared_ptr<cv::Mat>; using DetectResult = std::vector<DetectedObject>; // 自行定义,包含类别、置信度、框坐标 struct InferenceContext { rknn_context rknn_ctx; rknn_input inputs[1]; rknn_output outputs[1]; // 模型输入尺寸、类别数、anchor信息等 }; DetectResult runYolov5sFrame(InferenceContext* ctx, FramePtr frame) { // 1. 预处理:letterbox缩放 + 归一化 // 2. rknn_inputs_set + rknn_run + rknn_outputs_get // 3. 后处理:解码框、NMS // 4. 返回结果 } void inferenceLoop( ThreadPool* pool, BlockingQueue<FramePtr>* frameQueue, BlockingQueue<DetectResult>* resultQueue, InferenceContext* ctx) { while (true) { FramePtr frame; if (!frameQueue->pop(frame, std::chrono::milliseconds(100))) { // 队列超时就继续循环,顺带检查终止条件 continue; } auto task = pool->enqueue([ctx, frame]() -> DetectResult { return runYolov5sFrame(ctx, frame); }); // 这里不要阻塞等待task完成!否则线程池就被占满了。 // 实际做法是把future存到一个结果队列,由另一个线程去get。 resultQueue->push(std::make_shared<std::future<DetectResult>>(std::move(task))); } }

这里有个容易踩的深坑:不要在采集线程里enqueue任务后立刻get。如果这样写,采集线程会阻塞在get上,导致队列堆积、线程池里的线程全被占住。正确做法是把future本身也放进一个队列,让另一组线程去取结果。这也是为什么双路方案需要两个队列:帧队列解决“喂给谁”,结果队列解决“谁去收”。

3.4 双路实例化与安全的退出流程

双路方案在主函数里实例化两个独立的线程池,以及两组采集线程和结果队列。

int main() { // 路0 ThreadPool pool0(3); BlockingQueue<FramePtr> frameQueue0(16); // 有界队列,容量16 BlockingQueue<ResultPtr> resultQueue0(32); // 路1 ThreadPool pool1(3); BlockingQueue<FramePtr> frameQueue1(16); BlockingQueue<ResultPtr> resultQueue1(32); // 启动两路采集线程,分别往frameQueue0/frameQueue1放帧 std::thread capture0(captureLoop, /*cam0*/, &frameQueue0); std::thread capture1(captureLoop, /*cam1*/, &frameQueue1); // 启动两路推理管理线程,分别提交任务到pool0/pool1 std::thread infer0(inferenceLoop, &pool0, &frameQueue0, &resultQueue0, &ctx0); std::thread infer1(inferenceLoop, &pool1, &frameQueue1, &resultQueue1, &ctx1); // 主循环:从两个结果队列取数据、绘制、显示 while (running) { auto res0 = resultQueue0.pop(); auto res1 = resultQueue1.pop(); // 显示/业务逻辑 } running = false; capture0.join(); capture1.join(); infer0.join(); infer1.join(); return 0; }

退出流程的顺序很重要:先停止采集线程,再停推理管理线程,最后析构线程池。如果顺序反了,采集线程还在往池子里丢任务,线程池先被析构,enqueue就会抛异常。正确做法是先让采集线程退出,保证不再有新任务进来;然后让推理管理线程把队列里残留的帧处理完或直接清空;最后再等线程池析构。

4. 实测数据与踩坑实录:这些问题最容易翻车

4.1 用无界队列,内存翻倍还丢帧

双路方案第一次实测,我用的是无界队列,想当然地觉得“队列越大越不容易丢帧”。结果跑了几分钟,系统内存从300多MB一路涨到1GB以上,两路画面全都卡成幻灯片。

原因很清晰:当NPU推理速度小于摄像头帧率时,任务堆积在队列里。无界队列不限制容量,帧就一直在累积,每帧都是几MB的cv::Mat,系统内存很快就被吃光。而且堆积的帧几乎都是过期帧,处理它们毫无意义。

换成有界队列、容量16、满时丢弃最旧帧之后,内存稳定在400MB左右,画面延迟从数百毫秒降低到几十毫秒。这里最核心的认知是:视觉任务的队列不能用“可靠性”思路,而应该用“新鲜度”思路。宁可丢帧,不可积压。

4.2 推理线程从2加到4,FPS反而降了

我最初给每路配了4个推理线程,心想8核CPU多开几个线程总没坏处。实测打脸:每路FPS不仅没涨,反而从21帧左右掉到了17帧。

复盘原因:yolov5s的NPU推理是阻塞串行的,4个线程同时调rknn_run,NPU只能按顺序处理,线程大部分时间在等待。多出来的线程没有增加NPU吞吐,反而增加了调度开销和缓存竞争。RK3588的NPU处理一路yolov5s int8大约耗时10到15ms,单路实际瓶颈不在CPU而在NPU排队。

把每路推理线程池从4降到2,再配合预处理和后处理流水,FPS反而稳定在了22到25帧。两路合计接近50帧,已经能满足绝大多数场景。

4.3 CPU亲和性设置,避免采集和推理抢大核

双路跑起来之后还有个小问题:系统调度器有时把采集线程和推理线程同时放到A76大核上,A55小核反而闲着。结果大核被打满,小核空载,整体功耗还上去了。

解决方法是手动设置线程亲和性。采集线程绑到2个A55小核上,推理线程池的线程绑到2个A76大核上,主线程绑到剩下的大核。

void setThreadAffinity(std::thread& t, int cpuId) { cpu_set_t cpuset; CPU_ZERO(&cpuset); CPU_SET(cpuId, &cpuset); pthread_setaffinity_np(t.native_handle(), sizeof(cpu_set_t), &cpuset); }

RK3588的CPU编号一般是0-3为A55,4-7为A76。实测绑核之后,采集和推理之间互相干扰明显减少,两路帧率都更稳了。这个细节常规文档里很少写,但对双路方案的稳定性帮助很大。

4.4 双路帧同步:按时间戳取结果,不按等待

双路视觉最后要面临一个问题:两路画面不需要严格同步,但显示或业务处理时需要一个统一的“时刻”。如果主循环里先等resultQueue0,再等resultQueue1,某一路如果暂时没有结果,另一路也会被拖住。

我的处理方式是主循环以固定周期(比如33ms)去两个结果队列做非阻塞获取,各自取到最新结果就更新对应的显示区域,取不到就用上一帧的结果。两路之间不做强制关联,只保证每个结果都带时间戳,业务逻辑按时间戳判断是否过期。这样一路临时掉帧不会影响另一路的流畅度。

双路方案跑通的标志是什么?两路独立稳定运行、不互相拖累、内存不涨、帧率达标、任一路停止不影响另一路。做到这一步,RK3588双路视觉就已经从“能跑”跨到了“能用”。我自己调试下来最深的体会是:这套方案里线程池只是载体,真正决定成败的是对队列新鲜度和CPU资源分配的认知。后面如果做双路yolov8,或者把其中一路换成更高分辨率输入,这套“每路独立线程池+有界队列+亲和性绑定”的框架可以直接复用,只改模型上下文和预处理函数就行。

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

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

立即咨询