OpenCV结合threading实现摄像头帧环形缓冲区，get_last()取帧显示黑屏问题求助

核心故障点 #

两个关键错误直接导致黑屏结果：

1. 环形缓冲区的“最近帧”索引计算逻辑存在边界bug
   原代码的写入流程是：将读取到的帧写入self.buffer[self.current]后，再把self.current自增，到达max_size阈值后归零。但get_last()方法加了错误的边界判断——只有当self.current > 0时才把索引减1，当缓冲区写满一圈、self.current回到0的状态下，方法会直接取索引0的缓冲区内容，这位置存的是最早写入的旧帧（马上就要被新帧覆盖），完全不是最近捕获的帧。
2. 缺失OpenCV窗口渲染必需的事件循环调用
   OpenCV的GUI模块必须依靠waitKey()函数处理窗口消息、触发内容渲染，仅调用imshow()不接waitKey()的话，窗口会停留在初始黑屏状态，哪怕拿到的帧数据完全正常也无法显示。

具体修改方案 #

• 修正get_last()的索引计算逻辑，去掉错误的current > 0判断，用取模运算适配环形缓冲区的绕圈场景：
  错误原代码：

  def get_last(self):
      current = 0
      if self.current > 0:
          current = self.current - 1
      print('get_last()', current)
      return self.buffer[current]
  

  修正后代码：

  def get_last(self):
      # 环形缓冲区上一个写入位置的索引，自动适配绕圈的边界场景
      last_index = (self.current - 1) % self.max_size
      print('get_last()', last_index)
      return self.buffer[last_index]
  

• 补上cv2.waitKey()调用，让窗口可以完成渲染：
  原代码中imshow()之后直接用sleep等待，没有触发GUI事件处理，修改为：

  if frame is not None:
      print('showing frame')
      cv2.imshow('Frame', frame)
      # 阻塞3000毫秒处理窗口消息，期间窗口可以正常渲染帧内容
      cv2.waitKey(3000)
  

• （可选优化）给共享变量加线程锁，避免读写竞争：
  在__init__中初始化线程锁，对缓冲区的写入、读取操作加锁保护，防止偶发拿到半初始化的空帧。

修复后完整可运行代码 #

import cv2
import sys
import time
import threading

class CamBuffer(threading.Thread):
    def __init__(self, stream=0):
        self.current = 0
        self.max_size = 5
        self.buffer = [None] * self.max_size
        self.capture = cv2.VideoCapture(stream)
        # 新增线程锁保护共享变量
        self.lock = threading.Lock()
        if not self.capture.isOpened():
            print("Could not open video")
            sys.exit()

        print("Opened video")
        self.running = True
        super().__init__()

    def run(self):
        print('running thread')
        while self.running:
            print('capturing frame ', self.current)
            _, frame = self.capture.read()
            if _:
                print('saving frame ', self.current)
                with self.lock:
                    self.buffer[self.current] = frame
                    self.current = self.current + 1
                    if (self.current >= self.max_size):
                        self.current = 0
            # 加微小延时避免CPU空转
            time.sleep(0.01)
        self.capture.release()
        print('stopped thread')

    def terminate(self):
        print('terminating thread')
        self.running = False

    def get_last(self):
        with self.lock:
            last_index = (self.current - 1) % self.max_size
            print('get_last()', last_index)
            return self.buffer[last_index]
        

if __name__ == "__main__":
    print('Frame buffer test')
    stream = 0
    cb = CamBuffer(stream)
    cb.start()
    time.sleep(1.25)

    frame = cb.get_last()
    if frame is not None:
        print('showing frame')
        cv2.imshow('Frame', frame)
        cv2.waitKey(3000)

    cb.terminate()
    cb.join()
    cv2.destroyAllWindows()
    print('Frame buffer test [Done]')

内容的提问来源于stack exchange，提问作者Imran