PyQt6+OpenCV视频采集时窗口移动/缩放崩溃问题求助

核心问题分析 #

崩溃出现段错误，根源在于内存访问冲突和多线程资源竞争：

• QImage直接共享OpenCV帧的内存，当OpenCV释放帧内存后，QImage仍访问该内存导致非法内存访问
• 线程控制变量（is_running、is_paused）未做线程安全保护，多线程并发读写导致状态不一致
• GUI线程的paintEvent与视频线程的update_image并发访问pixmap，引发资源竞争
• 仅处理了窗口最小化事件，未覆盖调整大小、移动等高频交互场景

解决方案 #

1. 让QImage拥有独立内存

避免直接共享OpenCV的帧内存，创建QImage后立即复制数据，确保其拥有独立的内存空间：

def convert_cv_qt(self, cv_img):
    rgb_image = cv2.cvtColor(cv_img, cv2.COLOR_BGR2RGB)
    h, w, ch = rgb_image.shape
    bytes_per_line = ch * w
    # 复制数据，让QImage拥有独立内存
    return QImage(rgb_image.data, w, h, bytes_per_line, QImage.Format.Format_RGB888).copy()

2. 线程安全的状态控制

使用QAtomicBool替代普通布尔变量，确保多线程下状态读写的原子性，避免竞争：

from PyQt6.QtCore import QAtomicBool

class VideoThread(QThread):
    change_pixmap_signal = pyqtSignal(QImage)
    is_running = QAtomicBool(True)
    is_paused = QAtomicBool(False)

    def pause(self):
        self.is_paused.store(True)

    def resume(self):
        self.is_paused.store(False)

    def stop(self):
        self.is_running.store(False)
        self.quit()
        self.wait()

    def run(self):
        cap = cv2.VideoCapture(self.video_input)
        cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.resolution[0])
        cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.resolution[1])
        while self.is_running.load():
            if not self.is_paused.load():
                ret, cv_img = cap.read()
                if ret:
                    qt_img = self.convert_cv_qt(cv_img)
                    self.change_pixmap_signal.emit(qt_img)
            # 用QElapsedTimer更精确控制帧率，同时避免阻塞线程
            timer = QElapsedTimer()
            timer.start()
            while timer.elapsed() < 1000 / self.fps:
                if not self.is_running.load():
                    break
                QThread.msleep(1)
        cap.release()

3. 保护共享资源pixmap

使用QMutex确保pixmap的读写操作互斥，避免GUI线程和视频线程并发访问：

from PyQt6.QtCore import QMutex

class CameraApp(QWidget):
    def __init__(self):
        super().__init__()
        self.setGeometry(100, 100, 640, 480)
        self.initUI()
        self.pixmap = QPixmap()
        self.pixmap_mutex = QMutex()  # 新增互斥锁
        self.thread = VideoThread()
        self.thread.change_pixmap_signal.connect(self.update_image)
        self.thread.start()

    def paintEvent(self, event):
        self.pixmap_mutex.lock()
        pixmap_copy = self.pixmap.copy()
        self.pixmap_mutex.unlock()
        if not self.ispaused and not pixmap_copy.isNull():
            painter = QPainter(self)
            painter.drawPixmap(self.rect(), pixmap_copy)

    def update_image(self, qt_img):
        new_pixmap = QPixmap.fromImage(qt_img)
        self.pixmap_mutex.lock()
        self.pixmap = new_pixmap
        self.pixmap_mutex.unlock()
        self.update()

4. 全面处理窗口交互事件

覆盖调整大小、移动等事件，在交互期间暂停视频线程，避免高频交互时的资源冲突：

class CameraApp(QWidget):
    # ... 其他代码 ...
    def resizeEvent(self, event):
        self.thread.pause()
        super().resizeEvent(event)
        # 延迟恢复，避免频繁暂停/恢复
        QTimer.singleShot(100, self.thread.resume)

    def moveEvent(self, event):
        self.thread.pause()
        super().moveEvent(event)
        QTimer.singleShot(100, self.thread.resume)

    def event(self, e):
        if e.type() == QEvent.Type.WindowStateChange:
            if self.windowState() & Qt.WindowState.WindowMinimized:
                self.thread.pause()
                self.ispaused = True
            else:
                self.thread.resume()
                self.ispaused = False
        return super().event(e)

修改后的完整代码 #

import cv2
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QApplication
from PyQt6.QtCore import QThread, pyqtSignal, QEvent, Qt, QAtomicBool, QMutex, QElapsedTimer, QTimer
from PyQt6.QtGui import QImage, QPixmap, QPainter

class VideoThread(QThread):
    change_pixmap_signal = pyqtSignal(QImage)
    is_running = QAtomicBool(True)
    is_paused = QAtomicBool(False)

    def __init__(self, video_input=4, fps=30, resolution=(1920, 1080), parent=None):
        super().__init__(parent)
        self.fps = fps
        self.resolution = resolution
        self.video_input = video_input

    def run(self):
        cap = cv2.VideoCapture(self.video_input)
        cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.resolution[0])
        cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.resolution[1])
        while self.is_running.load():
            if not self.is_paused.load():
                ret, cv_img = cap.read()
                if ret:
                    qt_img = self.convert_cv_qt(cv_img)
                    self.change_pixmap_signal.emit(qt_img)
            # 精确控制帧率，同时响应停止信号
            timer = QElapsedTimer()
            timer.start()
            while timer.elapsed() < 1000 / self.fps:
                if not self.is_running.load():
                    break
                QThread.msleep(1)
        cap.release()

    def pause(self):
        self.is_paused.store(True)

    def resume(self):
        self.is_paused.store(False)

    def convert_cv_qt(self, cv_img):
        rgb_image = cv2.cvtColor(cv_img, cv2.COLOR_BGR2RGB)
        h, w, ch = rgb_image.shape
        bytes_per_line = ch * w
        # 复制数据，避免共享OpenCV内存
        return QImage(rgb_image.data, w, h, bytes_per_line, QImage.Format.Format_RGB888).copy()

    def stop(self):
        self.is_running.store(False)
        self.quit()
        self.wait()

class CameraApp(QWidget):
    ispaused = False
    def __init__(self):
        super().__init__()
        self.setGeometry(100, 100, 640, 480)
        self.initUI()
        self.pixmap = QPixmap()
        self.pixmap_mutex = QMutex()
        self.thread = VideoThread()
        self.thread.change_pixmap_signal.connect(self.update_image)
        self.thread.start()

    def initUI(self):
        self.layout = QVBoxLayout()
        self.setLayout(self.layout)

    def paintEvent(self, event):
        self.pixmap_mutex.lock()
        pixmap_copy = self.pixmap.copy()
        self.pixmap_mutex.unlock()
        if not self.ispaused and not pixmap_copy.isNull():
            painter = QPainter(self)
            painter.drawPixmap(self.rect(), pixmap_copy)

    def update_image(self, qt_img):
        new_pixmap = QPixmap.fromImage(qt_img)
        self.pixmap_mutex.lock()
        self.pixmap = new_pixmap
        self.pixmap_mutex.unlock()
        self.update()

    def resizeEvent(self, event):
        self.thread.pause()
        super().resizeEvent(event)
        QTimer.singleShot(100, self.thread.resume)

    def moveEvent(self, event):
        self.thread.pause()
        super().moveEvent(event)
        QTimer.singleShot(100, self.thread.resume)

    def event(self, e):
        if e.type() == QEvent.Type.WindowStateChange:
            if self.windowState() & Qt.WindowState.WindowMinimized:
                self.thread.pause()
                self.ispaused = True
            else:
                self.thread.resume()
                self.ispaused = False
        return super().event(e)

    def closeEvent(self, event):
        self.thread.stop()
        super().closeEvent(event)

if __name__ == "__main__":
    import sys
    app = QApplication(sys.argv)
    window = CameraApp()
    window.show()
    sys.exit(app.exec())

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