企业官网建设流程全解析

1. 双路视觉方案的整体设计思路

1.1 为什么要在RK3588上做双路视觉

单路摄像头跑yolov5s，在香橙派5上其实已经能跑得比较舒服了。RK3588这颗芯片带6TOPS算力的NPU，yolov5s量化成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 系统与基础依赖

我用的板子是香橙派5，16G内存版本，系统刷的是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/version

NPU驱动版本很关键，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，但做了两层封装：外层控制提交节奏，内层用有界队列加自定义拒绝策略。原因很简单——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_workers=PREPROCESS_WORKERS, thread_name_prefix="pre_" ) # IO专用池 io_pool = concurrent.futures.ThreadPoolExecutor( max_workers=IO_WORKERS, thread_name_prefix="io_" )

为什么把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_workers=max_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, interpolation=cv2.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, axis=0)

3.4 线程安全的帧缓冲设计

两路视频流共享线程池，帧数据在线程间传递，必须保证线程安全。我一开始用普通的list做缓冲，结果偶发数据错乱——一路的帧被另一路覆盖了。

后来改成每路独立的queue.Queue，配合帧ID做校验：

import queue import itertools class FrameBuffer: def __init__(self, maxsize=5): self.queue = queue.Queue(maxsize=maxsize) 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()

maxsize=5是权衡后的结果：太小容易丢帧，太大增加延迟。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, width=1920, height=1080, fps=30): super().__init__(daemon=True) 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(f"ch{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_ch=3配合4个工作线程，能保证线程池始终有活干，又不会堆积太多。

4.3 实测性能数据

在香橙派5上跑这个阶段一的骨架，两路1080P30输入，预处理到640x640，实测数据如下：

共享池的优势在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(f"task {task_id} start") try: return fn(*args, **kwargs) finally: logger.debug(f"task {task_id} done") self.semaphore.release()

日志一开，任务从提交到完成的全链路就清楚了。

5.2 摄像头读帧阻塞

现象：采集线程卡在cap.read()上，整个流水线停住。

原因通常是摄像头掉线或者驱动异常。OpenCV的read在设备异常时可能永久阻塞。解决办法是加超时机制，或者用V4L2的select/poll。

阶段一我用的简单方案：单独一个看门狗线程，定期检查采集线程的心跳，超过2秒没新帧就重启采集。

class CaptureWatchdog(threading.Thread): def __init__(self, capture, timeout=2.0): super().__init__(daemon=True) 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 = current

5.3 内存缓慢增长

现象：跑几个小时之后内存占用越来越高，最后OOM。

这种问题九成是引用没释放。Python的循环引用、numpy数组被闭包持有、future结果没清理，都会导致内存泄漏。

排查工具用tracemalloc：

import 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 常见问题速查表

实操心得：香橙派5的散热要注意，长时间满载CPU会降频。我加了个小风扇之后，预处理吞吐稳定了不少。如果发现跑一段时间性能下降，先摸一下散热片烫不烫。

6. 阶段一的验证与下一步衔接

6.1 怎么验证骨架是稳的

阶段一做完，我一般会跑一个24小时稳定性测试。两路摄像头持续输入，预处理任务持续提交，记录几个指标：帧率是否稳定、内存是否平稳、CPU温度是否可控、有没有异常日志。

验证脚本大概长这样：

import time import psutil def stability_test(pipeline, duration_hours=24): 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(interval=0.1) temp = psutil.sensors_temperatures() print(f"mem={mem}% cpu={cpu}% temp={temp}") time.sleep(0.001)

跑满24小时，内存波动在5%以内、帧率波动在10%以内，就算过关。

6.2 阶段二要接什么

阶段一的骨架搭好之后，阶段二要做的是把NPU推理接进来。具体包括：RKNN模型加载、双路推理的上下文管理、推理结果的异步回收、以及两路结果的融合逻辑。

这里有个关键决策：双路推理是串行还是并行。RK3588的NPU是单核的，两个模型实例同时跑会互相抢，所以大概率要串行化。但串行化之后，推理延迟会叠加，需要靠流水线并行来掩盖——也就是一路在推理的时候，另一路在做预处理，这样NPU不空闲。

这个调度逻辑，就是阶段二的核心。阶段一把线程池和缓冲做好，阶段二只需要在流水线里插入推理阶段，整体结构不用大改。

6.3 一些可以提前准备的事

如果你打算跟着这个系列往下做，阶段一结束之后可以先把这几件事做了：

• 把yolov5s模型转成RKNN格式，确认在香橙派5上能跑通单路
• 测一下单路推理的实际耗时，作为阶段二调度的依据
• 准备好两路摄像头的标定参数，如果涉及测距的话
• 把日志系统规范化，阶段二排查问题会依赖日志

我个人在实际操作中的体会是，双路视觉的难点从来不是模型本身，而是资源调度。线程池这个东西，配好了是利器，配不好就是灾难。阶段一慢一点、稳一点，后面会省很多事。