这个系列写到第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 contextNPU多核的优势也用不出来。内存带宽也需要计入成本。双路的MIPI CSI或USB摄像头取流加上两路图像的预处理每一帧RGBA或RGB888的数据量都不小。如果摄像头是4K分辨率一路就要吃掉不少内存带宽两路叠加后CPU缓存命中率下降线程调度延迟也会上升。这也是为什么线程池的线程数不能简单按“CPU核数1”来设。2.2 核心线程数、队列容量与拒绝策略的推荐值线程池关键参数就三个核心线程数、最大线程数、任务队列。以我的实践为准每路的线程池我建议这样配参数推荐值说明核心线程数2~3一路至少有2个推理线程并行yolov5s预处理和后处理可以流水最大线程数3不要超过4RK3588大核有限超过之后上下文切换开销大于收益任务队列类型有界队列容量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忙而暂时跑不动采集线程还能继续往队列放帧只要队列有界最坏情况就是丢弃旧帧。具体到代码结构我不建议用复杂的框架C11的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::functionvoid() task; { std::unique_lockstd::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::futuredecltype(f()) { using return_type decltype(f()); auto task std::make_sharedstd::packaged_taskreturn_type()( std::forwardF(f) ); std::futurereturn_type res task-get_future(); { std::unique_lockstd::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_lockstd::mutex lock(queue_mutex_); return tasks_.size(); } ~ThreadPool() { { std::unique_lockstd::mutex lock(queue_mutex_); stop_ true; } cv_.notify_all(); for (std::thread worker : workers_) { worker.join(); } } private: std::vectorstd::thread workers_; std::dequestd::functionvoid() tasks_; std::mutex queue_mutex_; std::condition_variable cv_; std::atomicbool stop_; };这个实现的几个关键点任务队列用std::deque而不是std::queue因为后面实现“丢弃最旧帧”需要从队头操作。wait的lambda条件是stop_或tasks_非空这样shutdown时线程能正常退出不会死等。packaged_task用来拿返回值便于上层知道推理任务是否完成或者拿到推理结果指针。析构时先置stop_再notify_all避免线程在等待中永远阻塞。3.3 把yolov5s的预处理、推理、后处理打包成任务双路方案里提交给线程池的不是一个“普通的函数”而是一个完整的推理任务。我把推理过程封装成一个lambda捕获帧数据和推理上下文指针。using FramePtr std::shared_ptrcv::Mat; using DetectResult std::vectorDetectedObject; // 自行定义包含类别、置信度、框坐标 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, BlockingQueueFramePtr* frameQueue, BlockingQueueDetectResult* 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_sharedstd::futureDetectResult(std::move(task))); } }这里有个容易踩的深坑不要在采集线程里enqueue任务后立刻get。如果这样写采集线程会阻塞在get上导致队列堆积、线程池里的线程全被占住。正确做法是把future本身也放进一个队列让另一组线程去取结果。这也是为什么双路方案需要两个队列帧队列解决“喂给谁”结果队列解决“谁去收”。3.4 双路实例化与安全的退出流程双路方案在主函数里实例化两个独立的线程池以及两组采集线程和结果队列。int main() { // 路0 ThreadPool pool0(3); BlockingQueueFramePtr frameQueue0(16); // 有界队列容量16 BlockingQueueResultPtr resultQueue0(32); // 路1 ThreadPool pool1(3); BlockingQueueFramePtr frameQueue1(16); BlockingQueueResultPtr 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加到4FPS反而降了我最初给每路配了4个推理线程心想8核CPU多开几个线程总没坏处。实测打脸每路FPS不仅没涨反而从21帧左右掉到了17帧。复盘原因yolov5s的NPU推理是阻塞串行的4个线程同时调rknn_runNPU只能按顺序处理线程大部分时间在等待。多出来的线程没有增加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为A554-7为A76。实测绑核之后采集和推理之间互相干扰明显减少两路帧率都更稳了。这个细节常规文档里很少写但对双路方案的稳定性帮助很大。4.4 双路帧同步按时间戳取结果不按等待双路视觉最后要面临一个问题两路画面不需要严格同步但显示或业务处理时需要一个统一的“时刻”。如果主循环里先等resultQueue0再等resultQueue1某一路如果暂时没有结果另一路也会被拖住。我的处理方式是主循环以固定周期比如33ms去两个结果队列做非阻塞获取各自取到最新结果就更新对应的显示区域取不到就用上一帧的结果。两路之间不做强制关联只保证每个结果都带时间戳业务逻辑按时间戳判断是否过期。这样一路临时掉帧不会影响另一路的流畅度。双路方案跑通的标志是什么两路独立稳定运行、不互相拖累、内存不涨、帧率达标、任一路停止不影响另一路。做到这一步RK3588双路视觉就已经从“能跑”跨到了“能用”。我自己调试下来最深的体会是这套方案里线程池只是载体真正决定成败的是对队列新鲜度和CPU资源分配的认知。后面如果做双路yolov8或者把其中一路换成更高分辨率输入这套“每路独立线程池有界队列亲和性绑定”的框架可以直接复用只改模型上下文和预处理函数就行。