1. 双路视觉方案的整体设计思路1.1 为什么要在RK3588上做双路视觉单路摄像头跑yolov5s在香橙派5上其实已经能跑得比较舒服了。RK3588这颗芯片带6TOPS算力的NPUyolov5s量化成INT8之后单帧推理时间大概在20到30毫秒之间算下来30到50帧的吞吐量应付一路1080P视频流绰绰有余。但实际项目里单路视觉往往不够用——比如要做双目测距、要做前视加后视的环视拼接、要做多工位质检或者像我这次接的一个小项目需要同时看两条传送带上的物料状态。这时候问题就来了两路摄像头同时采集如果各自开一套完整的推理流水线NPU会被两个进程争抢CPU也要同时扛两路解码和预处理内存带宽更是吃紧。我实测过最粗暴的方案——两个Python进程各跑各的yolov5s结果就是帧率直接掉到单路的六成左右而且延迟抖动特别大偶尔还会因为内存分配失败直接崩掉。所以双路视觉方案的核心矛盾不是“能不能跑”而是“怎么跑得稳、跑得省”。这就引出了这个阶段一的核心思路共享线程池。1.2 共享线程池到底共享的是什么很多人一听线程池第一反应是Java里的ThreadPoolExecutor或者Python里的concurrent.futures.ThreadPoolExecutor。概念是通的但放到RK3588的视觉流水线上需要想清楚一件事哪些任务适合放进池子哪些任务必须独占线程。我的拆解是这样的一路视觉流水线大致分成四个阶段——采集、解码、预处理、推理、后处理。其中采集和解码是IO密集型的预处理是CPU密集型的推理是NPU密集型的后处理是轻量CPU计算。如果两路各自维护一套线程那么峰值时刻会有大量线程同时抢CPU上下文切换的开销非常可观。共享线程池的思路是把两路流水线里同质化的CPU任务抽出来统一丢进一个池子里调度。比如两路的图像缩放、颜色空间转换、归一化这些预处理操作本质上是一样的计算完全可以共用一个工作线程池。NPU推理则通过RKNN的上下文管理来做串行化或者分时复用避免两个模型实例同时抢NPU。这样做的好处很直接线程数量可控CPU占用曲线更平滑内存分配更集中出问题的时候排查路径也清晰。1.3 阶段一要达成的目标这个教程是分阶段的阶段一不追求一步到位把双路全跑通而是先把共享线程池的骨架搭起来验证几个关键指标两路视频流能同时采集、解码不丢帧预处理任务能正确提交到共享池并回收结果单路推理链路在池化改造后性能不下降线程池的队列长度、拒绝策略、超时机制可观测把这几件事做扎实阶段二再接入双路NPU推理和结果融合就不会手忙脚乱。我见过太多人一上来就想把双路全打通结果卡在某个线程死锁上连日志都看不出来是哪一路出的问题。提示阶段一的所有代码建议先在PC上跑通逻辑再交叉编译到香橙派5上。RK3588的aarch64环境和x86在GIL表现、线程调度上差异不大但内存对齐和NEON指令相关的地方要特别留意。2. 香橙派5上的环境准备与关键配置2.1 系统与基础依赖我用的板子是香橙派516G内存版本系统刷的是Ubuntu 20.04 Server。选Server版是因为不需要桌面环境能省下不少内存和CPU给视觉任务。烧写过程用官方工具就行注意选对镜像版本我踩过一次坑早期某个版本的镜像里NPU驱动和RKNN Toolkit版本对不上导致模型加载直接报错。系统起来之后先做几件事# 更新源并安装基础依赖 sudo apt update sudo apt install -y python3-pip python3-dev cmake build-essential sudo apt install -y libopencv-dev python3-opencv sudo apt install -y v4l-utils ffmpeg # 检查NPU驱动版本 cat /sys/kernel/debug/rknpu/versionNPU驱动版本很关键RKNN Toolkit2的版本必须和它匹配。我这边驱动是0.9.2对应的RKNN Toolkit2用1.5.0以上比较稳。2.2 摄像头接入与MIPI配置双路视觉的采集端我这次用的是两路MIPI摄像头型号是OV5647和IMX219各一个。香橙派5有两个MIPI CSI接口但默认的设备树配置不一定两个都使能。需要改/boot/orangepiEnv.txt或者对应的dts overlay。# 查看当前video设备 ls /dev/video* # 用v4l2查看支持的格式 v4l2-ctl -d /dev/video0 --list-formats-ext v4l2-ctl -d /dev/video1 --list-formats-ext如果只看到video0没有video1大概率是设备树没配好。我当时的做法是参考官方wiki里的双摄overlay把两个CSI都打开。这里要注意两路同时跑1080P30的时候MIPI带宽是够的但如果上到4K就会开始丢帧所以阶段一我统一用1080P。2.3 Python线程池的选型考量Python里做线程池最直接的就是concurrent.futures.ThreadPoolExecutor。但视觉流水线有个特点任务提交频率高、单个任务耗时短、对延迟敏感。标准ThreadPoolExecutor的默认配置在这种场景下不一定合适。我对比过几种方案方案优点缺点适用场景ThreadPoolExecutor标准库、易用队列无界、拒绝策略弱任务量可控自建QueueWorker完全可控代码量大、易出错极致性能multiprocessing.Pool绕过GIL进程间通信开销大CPU密集且无共享状态asyncio线程高并发IO与OpenCV配合别扭纯IO场景最后我选的是ThreadPoolExecutor但做了两层封装外层控制提交节奏内层用有界队列加自定义拒绝策略。原因很简单——OpenCV的很多操作会释放GIL线程池在预处理阶段能真正并行起来而multiprocessing在传递图像数据时的序列化开销比省下的GIL时间还多。注意Python的GIL在NPU推理时不是瓶颈因为RKNN的推理调用会释放GIL。真正需要并行的是图像预处理这部分OpenCV的resize、cvtColor都会释放GIL所以线程池是划算的。3. 共享线程池的核心实现细节3.1 线程池参数怎么定线程池的大小不是拍脑袋定的。我的计算逻辑是这样的香橙派5的CPU是4个A76大核加4个A55小核。视觉流水线里采集和解码主要吃小核预处理吃大核推理走NPU。假设两路各需要1个采集线程、1个解码线程预处理是计算大头我给它分配4个工作线程那么总线程数控制在8到10之间比较合理。import concurrent.futures import os # 根据CPU核心数动态调整但设上限 CPU_COUNT os.cpu_count() # 8 PREPROCESS_WORKERS min(4, CPU_COUNT // 2) IO_WORKERS 2 # 预处理专用池 preprocess_pool concurrent.futures.ThreadPoolExecutor( max_workersPREPROCESS_WORKERS, thread_name_prefixpre_ ) # IO专用池 io_pool concurrent.futures.ThreadPoolExecutor( max_workersIO_WORKERS, thread_name_prefixio_ )为什么把IO和预处理分开因为IO任务经常阻塞如果和预处理混在一个池子里工作线程会被IO拖住导致预处理任务排队。分开之后IO池即使阻塞也不影响预处理池的吞吐。3.2 有界队列与背压机制ThreadPoolExecutor默认用的是无界队列任务提交速度超过处理速度时队列会无限增长最后OOM。视觉场景下这很危险——摄像头以30帧每秒往里灌如果预处理跟不上内存几分钟就爆了。我的做法是在提交层做背压维护一个信号量限制在途任务数量。import threading class BoundedExecutor: def __init__(self, max_workers, max_pending): self.pool concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) self.semaphore threading.Semaphore(max_pending) self.max_pending max_pending def submit(self, fn, *args, **kwargs): # 阻塞直到有空位实现背压 self.semaphore.acquire() future self.pool.submit(self._wrapper, fn, *args, **kwargs) return future def _wrapper(self, fn, *args, **kwargs): try: return fn(*args, **kwargs) finally: self.semaphore.release()max_pending设多少我的经验值是max_workers * 3。比如4个预处理线程在途任务上限12个。这样即使某一帧处理慢了也不会堆积太多丢帧比OOM好。3.3 任务粒度与批处理线程池的效率很大程度上取决于任务粒度。如果每个任务只是resize一张图任务太细调度开销占比高如果每个任务处理一整帧的所有预处理又可能单个任务太重导致负载不均。我试过三种粒度细粒度每个操作一个任务resize、cvtColor、normalize分开中粒度一帧的所有预处理作为一个任务粗粒度一批帧作为一个任务实测下来中粒度最合适。细粒度在香橙派5上调度开销能占到15%以上粗粒度则导致延迟增加。中粒度下单个任务耗时大概3到5毫秒线程池的调度开销可以忽略。def preprocess_frame(frame, target_size(640, 640)): 一帧的完整预处理作为一个任务提交 # resize resized cv2.resize(frame, target_size, interpolationcv2.INTER_LINEAR) # BGR转RGB rgb cv2.cvtColor(resized, cv2.COLOR_BGR2RGB) # 归一化并转NCHW normalized rgb.astype(np.float32) / 255.0 chw np.transpose(normalized, (2, 0, 1)) return np.expand_dims(chw, axis0)3.4 线程安全的帧缓冲设计两路视频流共享线程池帧数据在线程间传递必须保证线程安全。我一开始用普通的list做缓冲结果偶发数据错乱——一路的帧被另一路覆盖了。后来改成每路独立的queue.Queue配合帧ID做校验import queue import itertools class FrameBuffer: def __init__(self, maxsize5): self.queue queue.Queue(maxsizemaxsize) self.frame_id itertools.count() def put(self, frame): fid next(self.frame_id) try: self.queue.put_nowait((fid, frame)) except queue.Full: # 丢最旧的帧 try: self.queue.get_nowait() except queue.Empty: pass self.queue.put_nowait((fid, frame)) return fid def get(self): return self.queue.get()maxsize5是权衡后的结果太小容易丢帧太大增加延迟。5帧在30fps下大概是160毫秒的缓冲足够吸收短时抖动。实操心得帧ID一定要带上后处理阶段用它来对齐结果。我遇到过因为丢帧导致检测框和画面错位的问题加了帧ID校验之后一目了然。4. 双路采集与预处理的实操流程4.1 采集线程的实现每路摄像头一个采集线程用OpenCV的VideoCapture或者V4L2直接读。OpenCV简单但可控性差V4L2性能好但代码复杂。阶段一我用OpenCV因为要快速验证逻辑。import cv2 import threading class CameraCapture(threading.Thread): def __init__(self, device_id, buffer, width1920, height1080, fps30): super().__init__(daemonTrue) self.device_id device_id self.buffer buffer self.running True self.cap cv2.VideoCapture(device_id, cv2.CAP_V4L2) self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, width) self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height) self.cap.set(cv2.CAP_PROP_FPS, fps) # 减少内部缓冲降低延迟 self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) def run(self): while self.running: ret, frame self.cap.read() if not ret: continue self.buffer.put(frame) def stop(self): self.running False self.cap.release()CAP_PROP_BUFFERSIZE设成1很重要默认值可能导致读到的帧是几百毫秒前的。这个参数在V4L2后端下有效能明显降低延迟。4.2 预处理任务的提交与回收采集线程只管往缓冲里放帧预处理由主循环从缓冲取帧后提交到线程池。这里有个细节提交任务后不能立即等结果否则就变成串行了。我的做法是维护一个future列表批量提交、批量回收。class DualPipeline: def __init__(self, executor): self.executor executor self.buffers [FrameBuffer(), FrameBuffer()] self.pending {0: [], 1: []} self.max_pending_per_ch 3 def step(self): for ch in (0, 1): # 回收已完成的future done [f for f in self.pending[ch] if f.done()] for f in done: self.pending[ch].remove(f) try: result f.result() self.on_preprocess_done(ch, result) except Exception as e: print(fch{ch} preprocess error: {e}) # 提交新任务控制在途数量 while len(self.pending[ch]) self.max_pending_per_ch: try: fid, frame self.buffers[ch].get_nowait() except queue.Empty: break future self.executor.submit(preprocess_frame, frame) future.frame_id fid self.pending[ch].append(future)max_pending_per_ch3配合4个工作线程能保证线程池始终有活干又不会堆积太多。4.3 实测性能数据在香橙派5上跑这个阶段一的骨架两路1080P30输入预处理到640x640实测数据如下指标单路双路共享池双路独立池预处理吞吐280fps420fps380fpsCPU占用45%68%82%内存占用320MB410MB560MB帧延迟P9918ms26ms41ms共享池的优势在CPU占用和延迟抖动上很明显。独立池虽然吞吐略低但内存多用了150MB而且P99延迟翻倍说明线程争抢严重。注意这个数据是纯预处理阶段的还没接NPU推理。接上推理之后瓶颈会转移到NPU但共享池带来的CPU余量能让后处理更从容。5. 常见问题与排查技巧实录5.1 线程池任务丢失现象提交了任务但future永远不done或者结果对不上。排查思路先看线程池是否被shutdown了再看任务是否在队列里被拒绝。ThreadPoolExecutor在shutdown之后提交任务会直接抛RuntimeError但如果是队列满的情况标准库不会报错任务会静默排队。我的做法是给每个任务加唯一ID在wrapper里打日志def _wrapper(self, fn, *args, **kwargs): task_id next(self.task_counter) logger.debug(ftask {task_id} start) try: return fn(*args, **kwargs) finally: logger.debug(ftask {task_id} done) self.semaphore.release()日志一开任务从提交到完成的全链路就清楚了。5.2 摄像头读帧阻塞现象采集线程卡在cap.read()上整个流水线停住。原因通常是摄像头掉线或者驱动异常。OpenCV的read在设备异常时可能永久阻塞。解决办法是加超时机制或者用V4L2的select/poll。阶段一我用的简单方案单独一个看门狗线程定期检查采集线程的心跳超过2秒没新帧就重启采集。class CaptureWatchdog(threading.Thread): def __init__(self, capture, timeout2.0): super().__init__(daemonTrue) self.capture capture self.timeout timeout def run(self): last_count 0 while True: time.sleep(self.timeout) current self.capture.buffer.frame_id.__reduce__()[1][0] if current last_count: print(capture stalled, restarting...) self.capture.stop() self.capture.start() last_count current5.3 内存缓慢增长现象跑几个小时之后内存占用越来越高最后OOM。这种问题九成是引用没释放。Python的循环引用、numpy数组被闭包持有、future结果没清理都会导致内存泄漏。排查工具用tracemallocimport tracemalloc tracemalloc.start(10) # 跑一段时间后 snapshot tracemalloc.take_snapshot() top_stats snapshot.statistics(lineno) for stat in top_stats[:10]: print(stat)我遇到过一次是future对象一直存在pending列表里没清理因为done()判断有延迟。改成定期全量清理就好了。5.4 常见问题速查表问题可能原因排查方法解决帧率上不去线程池太小/任务太重看线程池活跃数调大max_workers或拆分任务延迟抖动大队列积压看在途任务数加背压、减小max_pending内存泄漏引用未释放tracemalloc清理future、断开循环引用采集卡死驱动异常看门狗心跳重启采集线程结果错位帧ID未校验打帧ID日志后处理对齐帧IDCPU占用高线程过多top -H减少线程数、合并任务实操心得香橙派5的散热要注意长时间满载CPU会降频。我加了个小风扇之后预处理吞吐稳定了不少。如果发现跑一段时间性能下降先摸一下散热片烫不烫。6. 阶段一的验证与下一步衔接6.1 怎么验证骨架是稳的阶段一做完我一般会跑一个24小时稳定性测试。两路摄像头持续输入预处理任务持续提交记录几个指标帧率是否稳定、内存是否平稳、CPU温度是否可控、有没有异常日志。验证脚本大概长这样import time import psutil def stability_test(pipeline, duration_hours24): start time.time() while time.time() - start duration_hours * 3600: pipeline.step() # 每10秒打一次状态 if int(time.time() - start) % 10 0: mem psutil.virtual_memory().percent cpu psutil.cpu_percent(interval0.1) temp psutil.sensors_temperatures() print(fmem{mem}% cpu{cpu}% temp{temp}) time.sleep(0.001)跑满24小时内存波动在5%以内、帧率波动在10%以内就算过关。6.2 阶段二要接什么阶段一的骨架搭好之后阶段二要做的是把NPU推理接进来。具体包括RKNN模型加载、双路推理的上下文管理、推理结果的异步回收、以及两路结果的融合逻辑。这里有个关键决策双路推理是串行还是并行。RK3588的NPU是单核的两个模型实例同时跑会互相抢所以大概率要串行化。但串行化之后推理延迟会叠加需要靠流水线并行来掩盖——也就是一路在推理的时候另一路在做预处理这样NPU不空闲。这个调度逻辑就是阶段二的核心。阶段一把线程池和缓冲做好阶段二只需要在流水线里插入推理阶段整体结构不用大改。6.3 一些可以提前准备的事如果你打算跟着这个系列往下做阶段一结束之后可以先把这几件事做了把yolov5s模型转成RKNN格式确认在香橙派5上能跑通单路测一下单路推理的实际耗时作为阶段二调度的依据准备好两路摄像头的标定参数如果涉及测距的话把日志系统规范化阶段二排查问题会依赖日志我个人在实际操作中的体会是双路视觉的难点从来不是模型本身而是资源调度。线程池这个东西配好了是利器配不好就是灾难。阶段一慢一点、稳一点后面会省很多事。