Python中如何将控制台打印内容高效追加到QPlainTextEdit控件

PyQt开发中需要将控制台所有print打印输出的内容实时更新显示到QPlainTextEdit控件，现有核心实现代码如下：

class Ui_logWindow(QtCore.QObject):

    logSignal = pyqtSignal(str)
    parsedLogSignal = pyqtSignal(str)
    # output_GUI = ''
    critical = True

    def showLogs(self):
        self.critical = True
        self.q = queue.Queue()
        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()
        self.client = ClientLogParser(arg1, arg2, arg3, arg4, arg5, arg6)
        self.client.client.on_message = self.onMsgWin
        self.logSignal.connect(self.appendLog)
        self.appendLogSimple = self.logSignal.emit
        self.parsedLogSignal.connect(self.appendParsedLog)
        self.appendLogParsed = self.parsedLogSignal.emit
        self.t = Thread(target = self.client.startListen, args=[])
        self.t.daemon = True
        self.t.start()
        self.t2 = Thread(target = self.processMsg, args=[])
        self.t2.daemon = True
        self.t2.start()

    def submit(self):
        self.showLogs()

    def appendLog(self,msg):
        self.log.appendPlainText(msg)

    def appendParsedLog(self, msg):
        self.parsedLog.appendPlainText(msg)

    def onMsgWin(self, client, userdata, msg):
        self.listMsg = msg.payload.decode('utf-8').rstrip()
        self.q.put(self.listMsg)
        
    def processMsg(self):
        while self.critical == True:
            try:
                self.listMsgProcessed = self.q.get(self.listMsg)
            except:
                sleep(.1)
            else:                
                self.appendLogSimple(self.listMsgProcessed)
                for line in self.listMsgProcessed.split('\n'):
                    self.client.lp.parseEvent(line)
                    self.client.rp.evaluateDict(self.client.lp.dictLogs)
                    # self.output_GUI += "START\n"
                    print('Start')
                    for key, val in self.client.lp.dictLogs.items():
                        print(f"EVT : [{key}]")
                        # self.output_GUI += f"EVT : [{key}]"
                        if val == 1 :
                            # self.output_GUI += f'[{key}] has been triggered [{val}] time \n'
                            print(f'[{key}] has been triggered [{val}] time')
                        elif val > 1:
                            # self.output_GUI += f'[{key}] has been triggered [{val}] times \n'
                            print(f'[{key}] has been triggered [{val}] times')
                    print("RULES:")
                    # self.output_GUI += "RULES:\n"
                    for rule_idx in range(len(self.client.rp.dictRulesList)):
                        print(f"RULE : [{rule_idx}] [{self.client.rp.dictRulesList[rule_idx]}]")
                        # self.output_GUI += f"RULE : [{rule_idx}] [{self.client.rp.dictRulesList[rule_idx]}]\n "
                        if self.client.rp.resultList[rule_idx] == 'warning':
                            print(f"RESULT : [{rule_idx}] action changed [{self.client.rp.resultList[rule_idx]}]")
                            # self.output_GUI += f"RESULT : [{rule_idx}] action changed [{self.client.rp.resultList[rule_idx]}]\n"
                        if self.client.rp.resultList[rule_idx] == 'info':
                            print(f"RESULT : [{rule_idx}] action changed [{self.client.rp.resultList[rule_idx]}]")
                            # self.output_GUI += f"RESULT : [{rule_idx}] action changed [{self.client.rp.resultList[rule_idx]}]\n"
                        if self.client.rp.resultList[rule_idx] == 'succes':
                            print(f"RESULT : [{rule_idx}] action changed [{self.client.rp.resultList[rule_idx]}]")
                            # self.output_GUI += f"RESULT : [{rule_idx}] action changed [{self.client.rp.resultList[rule_idx]}]\n"
                        if self.client.rp.resultList[rule_idx] == 'critical':
                            print(f"RESULT : [{rule_idx}] action changed [{self.client.rp.resultList[rule_idx]}]")
                            # self.output_GUI += f"RESULT : [{rule_idx}] action changed [{self.client.rp.resultList[rule_idx]}]\n"
                            print(f'Action critical triggered.')
                            # self.output_GUI += f'Action critical triggered.\n'
                            # self.appendLogParsed(self.output_GUI)
                            self.critical = False
                    print('End')
                    # self.output_GUI += "END \n"
                    # self.appendLogParsed(self.output_GUI)

    def setupUi(self, logWindow):
        self.log = QtWidgets.QPlainTextEdit(self.centralwidget)
        self.log.setGeometry(QtCore.QRect(430, 20, 1250, 311))
        self.log.setReadOnly(True)
        self.log.setObjectName("plainTextEdit_2")
        self.log.setMaximumBlockCount(15)
        self.parsedLog = QtWidgets.QPlainTextEdit(self.centralwidget)
        self.parsedLog.setGeometry(QtCore.QRect(20, 350, 1660, 640))
        self.parsedLog.setReadOnly(True)
        self.parsedLog.setObjectName("plainTextEdit")
        self.parsedLog.setMaximumBlockCount(50)

当前实现中，显示原始日志的self.log控件运行正常，但processMsg方法内的所有print输出无法高效追加到self.parsedLog控件。此前尝试定义output_GUI字符串变量逐行拼接打印内容后再追加到控件，在高速日志输入场景下性能极差、界面卡顿严重，需要可落地的高性能实现方案。

核心思路是直接重定向标准输出流，替代手动字符串拼接的方案，结合Qt信号槽的跨线程队列机制保证线程安全，同时给文本控件开启高性能渲染配置降低开销。
注意：Qt所有UI控件操作必须在主线程执行，禁止在工作线程直接调用appendPlainText等UI方法，否则会出现随机崩溃、界面无响应问题。

具体实现步骤 #

• 第一步：自定义标准输出重定向类，拦截所有print输出，通过Qt信号投递内容，同时保留原有控制台输出能力，不影响终端调试：

import sys
from io import TextIOBase
from PyQt5.QtCore import QObject, pyqtSignal

class StdoutRedirect(QObject, TextIOBase):
    logEmitted = pyqtSignal(str)
    def write(self, content):
        # 过滤空字符串，避免无效UI刷新
        if content and not content.isspace():
            self.logEmitted.emit(content)
        # 同步输出到原始终端
        return sys.__stdout__.write(content)
    
    def flush(self):
        sys.__stdout__.flush()

• 第二步：在日志窗口初始化时绑定重定向逻辑，替换系统标准输出，将重定向信号连接到日志追加方法：

def showLogs(self):
    self.critical = True
    self.q = queue.Queue()
    # 初始化重定向实例，避免重复绑定
    if not hasattr(self, '_stdout_redirect'):
        self._stdout_redirect = StdoutRedirect()
        self._stdout_redirect.logEmitted.connect(self.appendParsedLog)
    # 替换系统标准输出
    sys.stdout = self._stdout_redirect

    # 原有业务初始化逻辑保留，不需要修改
    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()
    self.client = ClientLogParser(arg1, arg2, arg3, arg4, arg5, arg6)
    self.client.client.on_message = self.onMsgWin
    self.logSignal.connect(self.appendLog)
    self.appendLogSimple = self.logSignal.emit
    self.t = Thread(target = self.client.startListen, args=[])
    self.t.daemon = True
    self.t.start()
    self.t2 = Thread(target = self.processMsg, args=[])
    self.t2.daemon = True
    self.t2.start()

• 第三步：删除原代码中所有output_GUI相关的字符串拼接逻辑，processMsg方法内的print语句不需要做任何修改，所有输出会自动同步到self.parsedLog控件。

• 第四步：给QPlainTextEdit增加高性能渲染配置，降低高速日志下的渲染开销：

def setupUi(self, logWindow):
    self.log = QtWidgets.QPlainTextEdit(self.centralwidget)
    self.log.setGeometry(QtCore.QRect(430, 20, 1250, 311))
    self.log.setReadOnly(True)
    self.log.setObjectName("plainTextEdit_2")
    self.log.setMaximumBlockCount(15)
    self.parsedLog = QtWidgets.QPlainTextEdit(self.centralwidget)
    self.parsedLog.setGeometry(QtCore.QRect(20, 350, 1660, 640))
    self.parsedLog.setReadOnly(True)
    self.parsedLog.setObjectName("plainTextEdit")
    self.parsedLog.setMaximumBlockCount(50)
    # 新增高性能配置
    self.parsedLog.setLineWrapMode(QtWidgets.QPlainTextEdit.NoWrap)
    self.parsedLog.setCenterOnScroll(False)
    self.parsedLog.document().setUndoRedoEnabled(False)

方案优势 #

• 零侵入接入：不需要修改原有业务代码中的print语句，接入成本极低
• 性能优异：不存在大字符串反复拼接、拷贝的内存开销，每一行日志直接通过信号投递，内存占用稳定
• 线程安全：利用Qt信号槽自带的跨线程队列连接机制，自动保证UI操作在主线程执行，不会出现跨线程操作UI的崩溃问题
• 兼容原有调试流程：打印内容会同步输出到系统终端，不影响命令行调试

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