PyQt5相机日志解析程序偶发Segmentation fault崩溃求助

你遇到的随机段错误（Segmentation fault (core dumped)）核心原因是违反了Qt的线程安全规则：

• Qt框架明确要求所有GUI控件的操作必须在主线程（Qt事件循环运行的线程）执行，跨线程直接调用控件方法会触发底层C++对象的内存访问冲突，表现就是随机崩溃——日志生成速率越快，跨线程GUI操作越频繁，崩溃概率越高，和你加deque缓存、sleep延时没有关系。
• 你代码里的onMsgWin是MQTT客户端的消息回调，运行在自己启动的后台监听线程中，你在这个回调里直接调用self.textEdit.append()、self.textEdit.moveCursor()、self.textEdit_2.append()操作UI控件，就是崩溃的直接诱因。
• 代码里的其他问题会进一步提升崩溃概率：
  • 子线程中直接调用sys.exit(-1)会绕过Qt的资源回收流程，强制终止进程时极易触发段错误
  • 后台线程和主线程同时读写self.client.lp.dictLogs、self.client.rp.resultList等共享变量没有加锁，存在数据竞争，会导致内存访问异常
  • 队列长度达到20后你在消息回调线程里sleep(5)，会直接阻塞MQTT消息接收循环，导致消息堆积、内存占用异常升高
  • 队列长度小于20时，每新增一行日志你就遍历整个队列重复计算统计结果、重复拼接输出内容，会产生大量无效的重复GUI更新操作，进一步提升跨线程冲突概率

• 用Qt自带的信号槽机制做跨线程数据传递，后台线程只负责接收日志、做解析计算，绝对不能直接操作任何GUI控件；所有UI更新操作全部通过信号触发，由主线程执行，Qt信号槽会自动处理跨线程的线程安全问题。
• 移除消息回调里的sleep逻辑，不要阻塞MQTT消息接收循环；deque可以直接指定maxlen=20，不用手动判断长度做清空。
• 移除子线程里的sys.exit调用，critical错误通过信号通知主线程做正常的资源回收和退出。
• 优化统计逻辑，不要每加一行就重复遍历整个队列计算统计结果，队列攒满20行计算一次输出即可，减少无效计算和UI更新。

修正后的核心代码参考：

from collections import deque
import sys
import os
from threading import Thread
from time import sleep

sys.path.append(os.path.abspath(os.path.dirname(os.path.dirname('../'))))

from livelogparser.clientlogParser import ClientLogParser 
from PyQt5 import QtCore, QtWidgets
from PyQt5.QtCore import QThread, QObject, pyqtSignal
from PyQt5.QtGui import QTextCursor

ON_POSIX = 'posix' in sys.builtin_module_names

class Ui_logWindow(object):
    # 定义跨线程信号，分别传递原始日志、统计结果、致命错误通知
    sig_raw_log = pyqtSignal(str)
    sig_stat_log = pyqtSignal(str)
    sig_critical_error = pyqtSignal()

    def setupUi(self, logWindow):
        # 保留原有UI初始化逻辑，新增信号和槽的绑定
        # ... 你原有lineEdit、textEdit、textEdit_2等控件初始化代码 ...
        self.sig_raw_log.connect(self._append_raw_log)
        self.sig_stat_log.connect(self._append_stat_log)
        self.sig_critical_error.connect(self._handle_critical_exit)

    def _append_raw_log(self, content):
        # 仅在主线程执行，操作原始日志控件
        self.textEdit.append(content)
        self.textEdit.moveCursor(QTextCursor.End)

    def _append_stat_log(self, content):
        # 仅在主线程执行，操作统计结果控件
        self.textEdit_2.append(content)
        self.textEdit_2.moveCursor(QTextCursor.End)

    def _handle_critical_exit(self):
        # 仅在主线程执行退出逻辑，避免资源泄漏
        QtWidgets.QMessageBox.critical(None, "致命错误", "检测到critical级别规则触发，程序即将退出")
        QtWidgets.QApplication.quit()

    def logs(self):
        arg1 = self.lineEdit.text()
        arg2 = int(self.lineEdit_2.text())
        arg3 = self.lineEdit_3.text()
        arg4 = self.lineEdit_4.text()
        arg5 = self.lineEdit_5.text()
        arg6 = self.lineEdit_6.text()
        # 直接指定deque最大长度，无需手动判断溢出
        self.q = deque(maxlen=20)
        self.client = ClientLogParser(arg1, arg2, arg3, arg4, arg5, arg6)
        self.client.client.on_message = self.onMsgWin
        self.t = Thread(target=self.client.startListen, args=[], daemon=True)
        self.t.start()

    def onMsgWin(self, client, userdata, msg):
        # 该函数运行在后台子线程，绝对不能直接操作GUI控件
        listMsg = msg.payload.decode('utf-8').rstrip()
        # 通过信号传递原始日志到主线程更新UI
        self.sig_raw_log.emit(listMsg)
        
        for line in listMsg.split('\n'):
            self.client.lp.parseEvent(line)
            self.q.append(line)
            # 队列攒满20行再做一次统计输出，避免重复计算
            if len(self.q) == 20:
                output_GUI = ''
                self.client.rp.evaluateDict(self.client.lp.dictLogs)
                output_GUI += "COUNT:\n"
                for key, val in self.client.lp.dictLogs.items():
                    output_GUI += f"EVT[{key}] COUNT[{val}]\n"
                    if val > 0:
                        output_GUI += f'------ the event : {key} has been triggered ------\n'
                output_GUI += "RULES:\n"
                hit_critical = False
                for rule_idx in range(len(self.client.rp.dictRulesList)):
                    res = self.client.rp.resultList[rule_idx]
                    if res == 'critical':
                        hit_critical = True
                        break
                    if res in ('warning', 'info', 'succes'):
                        output_GUI += '----------succes----------\n'
                    output_GUI += f"RULE:[{rule_idx}][{self.client.rp.dictRulesList[rule_idx]}]\n"
                    output_GUI += f"RESULT[{rule_idx}][{res}]\n"
                
                if hit_critical:
                    self.sig_critical_error.emit()
                    return
                
                output_GUI += "----ITERATION-END----\n"
                # 通过信号传递统计结果到主线程更新UI
                self.sig_stat_log.emit(output_GUI)
                self.q.clear()

额外优化建议 #

如果后续日志解析逻辑耗时变长，建议把解析逻辑也拆分到单独的工作线程，解析完成后再发信号更新UI，避免阻塞MQTT消息接收；如果出现偶发的统计数据错乱，加一把threading.Lock保护共享字典、规则列表的读写操作即可。

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