09-多线程推理架构:图像采集、推理、显示解耦

📅 2026/8/10 2:28:14
09-多线程推理架构:图像采集、推理、显示解耦
多线程推理架构:图像采集、推理、显示解耦为什么需要多线程前面几篇的代码都是单线程的:读一帧 → 预处理 → 推理 → 后处理 → 显示 → 读下一帧。这些步骤串行执行,总耗时是各步骤时间之和。假设各环节耗时:读取 5ms + 预处理 3ms + 推理 60ms + 后处理 5ms + 显示 3ms =76ms/帧 ≈ 13FPS。问题在哪?推理那 60ms 里,CPU 在干活,但摄像头采集和显示都在空等。摄像头明明能 30FPS 出帧,但你 13FPS 去读,中间的帧全浪费了。多线程的核心思路:把采集、推理、显示拆成独立线程,各干各的,通过队列传递数据。推理线程忙着的时候,采集线程继续读帧攒着,显示线程画上一次的结果,互不阻塞。多线程架构设计┌─────────────┐ Queue ┌──────────────┐ Queue ┌──────────────┐ │ 采集线程 │──────────────→│ 推理线程 │─────────────→│ 显示线程 │ │ capture() │ latest frame │ inference() │ results │ display() │ └─────────────┘ └──────────────┘ └──────────────┘ │ │ │ │ 持续读摄像头 │ 持续推理 │ 持续刷新画面 └──────────────────────────────────────────────────────────────┘三个线程各司其职:采集线程:死循环读摄像头帧,把最新帧塞进队列,队列满了就丢旧帧推理线程:从队列取帧,预处理 → 推理 → 后处理,把结果塞进结果队列显示线程:从结果队列取检测结果,画框显示,同时显示最新采集帧完整代码实现importcv2importnumpyasnpimportonnxruntimeasortimportthreadingimportqueueimporttimeclassFrameQueue:"""线程安全的帧队列,只保留最新帧"""def__init__(self,maxsize=2):self.queue=queue.Queue(maxsize=maxsize)self.lock=threading.Lock()self.latest=None# 始终保留最新帧defput(self,frame):withself.lock:self.latest=frame# 队列满了就丢弃旧帧,不阻塞try:self.queue.put_nowait(frame)exceptqueue.Full:passdefget(self,timeout=1.0):try:returnself.queue.get(timeout=timeout)exceptqueue.Empty:withself.lock:returnself.latest# 队列空了就用最新帧defget_latest(self):withself.lock:returnself.latestclassResultQueue:"""检测结果队列"""def__init__(self,maxsize=3):self.queue=queue.Queue(maxsize=maxsize)self.latest=Noneself.lock=threading.Lock()defput(self,result):withself.lock:self.latest=resulttry:self.queue.put_nowait(result)exceptqueue.Full:passdefget_latest(self):withself.lock:returnself.latestclassCaptureThread(threading.Thread):"""采集线程:持续读取摄像头帧"""def__init__(self,source,frame_queue):super().__init__(daemon=True)self.source=source self.frame_queue=frame_queue self.running=Trueself.fps_counter=FPSCounter()defrun