QThread

在 GUI 程序中，如果想要让程序执行一项耗时的工作，例如下载文件、I/O 存取等,深度学习模型推理，直接在 UI 线程中进行这些操作会导致整个界面卡住，出现无响应的状态。为了避免这种情况，可以将这些耗时任务放在另一个线程中执行。在 PyQt 中，可以使用 QThread 来实现这一点。

基本用法

在 PyQt5 中，要使用 QThread 创建一个线程，需要创建 QThread 的子类，并重写 QThread.run() 函数。以下是一个示例，WorkerThread 类继承自 QThread 并重写了 run() 方法：

class WorkerThread(QThread): def __init__(self): super().__init__() def run(self): # 执行耗时任务

接下来，可以使用 QThread.start() 来启动线程。在构造 WorkerThread 后，不会立即执行 run() 方法，直到调用 QThread.start() 才开始执行。如果主线程需要等待线程执行完毕，可以使用 QThread.wait()，它会等待线程完成才返回。但请注意，如果线程中包含无限循环，QThread.wait() 将会无限期等待。

下面是一个简单的示例：

import sysimport timefrom PyQt5.QtWidgets import QApplicationfrom PyQt5.QtCore import QThreadclass WorkerThread(QThread): def __init__(self): super().__init__() def run(self): for i in range(3): time.sleep(1) print('WorkerThread::run ' + str(i))if __name__ == '__main__': app = QApplication(sys.argv) print('main') work1 = WorkerThread() work2 = WorkerThread() work1.start() work2.start() work1.wait() work2.wait() print('end of main') sys.exit(app.exec_())

在这个 run 函数中，有一个循环执行 3 次，并在每次循环后休眠 1 秒，然后输出一条消息。从输出结果可以看出，主线程会等待两个 WorkerThread 线程都完成后才输出 “end of main” 消息，这证明了使用 QThread.wait() 是有效的。

错误堵塞机制

在 PyQt5 开发过程中，新手容易错误地使用线程，导致界面卡住或变黑。例如，一个下载文件的程序，按下按钮后会执行大约 10 秒的工作。如果在 UI 线程中直接执行这些操作，会发现整个 GUI 程序无法进行其他操作，界面会卡住或变黑。

正确的做法是在 WorkerThread 中定义两个信号，分别为 trigger 和 finished。finished 信号在工作完成后发送，而 trigger 信号在执行过程中发送。使用 pyqtSignal 来定义自定义信号，并指定发送到目标函数的参数类型。例如，trigger = pyqtSignal(str)。

整个程序在按下按钮后，会开启另一个线程，每秒更新一次秒数到标签上，通过 self.trigger.emit(str(i+1)) 发射信号并传递第几秒的参数。第 5 秒时结束线程，并通过 self.finished.emit() 发射结束信号。

下面是一个示例代码：

import sysimport timefrom PyQt5.QtWidgets import QApplication, QWidget, QVBoxLayout, QLabel, QPushButtonfrom PyQt5.QtCore import QThread, pyqtSignalclass WorkerThread(QThread): trigger = pyqtSignal(str) finished = pyqtSignal() def __init__(self): super().__init__() def run(self): for i in range(5): time.sleep(1) self.trigger.emit(str(i+1)) print('WorkerThread::run ' + str(i)) self.finished.emit()class MyWidget(QWidget): def __init__(self): super().__init__() self.initUI() def initUI(self): self.setWindowTitle('my window') self.setGeometry(50, 50, 200, 150) layout = QVBoxLayout() self.setLayout(layout) self.mylabel = QLabel('press button to start thread', self) layout.addWidget(self.mylabel) self.mybutton = QPushButton('start', self) self.mybutton.clicked.connect(self.startThread) layout.addWidget(self.mybutton) self.work = WorkerThread() def startThread(self): self.mybutton.setDisabled(True) self.work.start() self.work.trigger.connect(self.updateLabel) self.work.finished.connect(self.threadFinished) self.updateLabel(str(0)) def threadFinished(self): self.mybutton.setDisabled(False) def updateLabel(self, text): self.mylabel.setText(text)if __name__ == '__main__': app = QApplication(sys.argv) w = MyWidget() w.show() sys.exit(app.exec_())

结果如下：

mainpress button to start thread1234end of main

如果不想在 threadFinished 函数中只是执行一段代码，可以省略 threadFinished 函数，改用 lambda 表达式。例如，将 self.mybutton.setDisabled(False) 操作写在 self.work.finished.connect() 中的 lambda 表达式里。

在 QWidget 里使用 QThread

在 PyQt 程序中，主线程就是所说的 UI 线程，UI 线程会处理所有控件的事务。因此，如果有耗时的工作需要执行，通常不会将其放在 UI 线程中，因为这样做会阻止其他控件的更新，导致界面卡顿或程序无响应。解决这个问题的方法是创建另一个线程来处理这些耗时的工作。

在 WorkerThread 中新增了两个信号，分别命名为 trigger 和 finished。finished 信号在工作完成后发送，而 trigger 信号在执行过程中发送。当需要自定义信号时，使用 pyqtSignal 来定义要发送到目标函数的函数原型，例如在下面的示例中，trigger = pyqtSignal(str) 表示 trigger 信号会携带一个字符串参数。

整个程序的工作流程是：当按下按钮后，会启动另一个线程，这个线程每一秒更新一次秒数到标签上。通过 self.trigger.emit(str(i+1)) 发送 trigger 信号，并传递当前的秒数作为参数。当到达第 5 秒时，线程结束，并通过 self.finished.emit() 发送 finished 信号。

这种设计允许我们在不影响 UI 响应性的情况下执行长时间的任务，同时还能更新 UI 显示当前的进度或状态。通过使用信号和槽机制，可以在工作线程和 UI 线程之间安全地传递信息，确保程序的流畅运行。

#!/usr/bin/env python3# -*- coding: utf-8 -*-import sysimport timefrom PyQt5.QtWidgets import (QApplication, QWidget, QVBoxLayout, QLabel, QPushButton)from PyQt5.QtCore import QThread, pyqtSignalclass WorkerThread(QThread): trigger = pyqtSignal(str) finished = pyqtSignal() def __init__(self): super().__init__() def run(self): for i in range(5): time.sleep(1) self.trigger.emit(str(i+1)) print('WorkerThread::run ' + str(i)) self.finished.emit()class MyWidget(QWidget): def __init__(self): super().__init__() self.initUI() def initUI(self): self.setWindowTitle('my window') self.setGeometry(50, 50, 200, 150) layout = QVBoxLayout() self.setLayout(layout) self.mylabel = QLabel('press button to start thread', self) layout.addWidget(self.mylabel) self.mybutton = QPushButton('start', self) self.mybutton.clicked.connect(self.startThread) layout.addWidget(self.mybutton) self.work = WorkerThread() def startThread(self): self.mybutton.setDisabled(True) self.work.start() self.work.trigger.connect(self.updateLabel) self.work.finished.connect(self.threadFinished) self.updateLabel(str(0)) def threadFinished(self): self.mybutton.setDisabled(False) def updateLabel(self, text): self.mylabel.setText(text)if __name__ == '__main__': app = QApplication(sys.argv) w = MyWidget() w.show() sys.exit(app.exec_())

如果你不希望 threadFinished 函数仅仅是执行一行代码，而是想去掉 threadFinished 函数，可以将 self.mybutton.setDisabled(False) 操作写在 self.work.finished.connect() 里的 lambda 表达式中。以下是使用 lambda 表达式的示例：

def startThread(self): self.mybutton.setDisabled(True) self.work.start() self.work.trigger.connect(self.updateLabel) # self.work.finished.connect(self.threadFinished) self.work.finished.connect(lambda: self.mybutton.setDisabled(False)) self.updateLabel(str(0))# def threadFinished(self):# self.mybutton.setDisabled(False)def updateLabel(self, text): self.mylabel.setText(text)

这样，当 finished 信号被触发时，lambda 表达式中的代码将会执行，即启用按钮。