在 PyQt 中正确使用 QThread 的示例？

问题描述：

我正在尝试学习如何在 PyQt Gui 应用程序中使用 QThreads。我有一些运行了一段时间的东西，（通常）有一些可以更新 Gui 的点，但我想将主要工作拆分到它自己的线程中（有时东西会卡住，最终有一个取消/重试按钮会很好，如果 Gui 因主循环被阻塞而冻结，这显然不起作用）。

我读过https://mayaposch.wordpress.com/2011/11/01/how-to-really-truly-use-qthreads-the-full-explanation/。该页面说重新实现该run方法不是解决问题的方法。我遇到的问题是找到一个 PyQt 示例，其中有一个主线程执行 Gui，而一个工作线程不这样做。这篇博文是针对 C++ 的，所以虽然它的例子确实有帮助，但我仍然有点迷茫。有人能给我指出一个在 Python 中正确执行此操作的示例吗？

解决方案 1：

这是一个单独的工作线程的工作示例，它可以发送和接收信号以允许它与 GUI 通信。

我制作了两个简单的按钮，一个在单独的线程中启动长时间的计算，另一个立即终止计算并重置工作线程。

像这里所做的那样强制终止线程通常不是最好的做法，但在某些情况下，总是正常退出并不是一个选择。

from PyQt4 import QtGui, QtCore
import sys
import random

class Example(QtCore.QObject):

    signalStatus = QtCore.pyqtSignal(str)

    def __init__(self, parent=None):
        super(self.__class__, self).__init__(parent)

        # Create a gui object.
        self.gui = Window()

        # Create a new worker thread.
        self.createWorkerThread()

        # Make any cross object connections.
        self._connectSignals()

        self.gui.show()


    def _connectSignals(self):
        self.gui.button_cancel.clicked.connect(self.forceWorkerReset)
        self.signalStatus.connect(self.gui.updateStatus)
        self.parent().aboutToQuit.connect(self.forceWorkerQuit)


    def createWorkerThread(self):

        # Setup the worker object and the worker_thread.
        self.worker = WorkerObject()
        self.worker_thread = QtCore.QThread()
        self.worker.moveToThread(self.worker_thread)
        self.worker_thread.start()

        # Connect any worker signals
        self.worker.signalStatus.connect(self.gui.updateStatus)
        self.gui.button_start.clicked.connect(self.worker.startWork)


    def forceWorkerReset(self):      
        if self.worker_thread.isRunning():
            print('Terminating thread.')
            self.worker_thread.terminate()

            print('Waiting for thread termination.')
            self.worker_thread.wait()

            self.signalStatus.emit('Idle.')

            print('building new working object.')
            self.createWorkerThread()


    def forceWorkerQuit(self):
        if self.worker_thread.isRunning():
            self.worker_thread.terminate()
            self.worker_thread.wait()


class WorkerObject(QtCore.QObject):

    signalStatus = QtCore.pyqtSignal(str)

    def __init__(self, parent=None):
        super(self.__class__, self).__init__(parent)

    @QtCore.pyqtSlot()        
    def startWork(self):
        for ii in range(7):
            number = random.randint(0,5000**ii)
            self.signalStatus.emit('Iteration: {}, Factoring: {}'.format(ii, number))
            factors = self.primeFactors(number)
            print('Number: ', number, 'Factors: ', factors)
        self.signalStatus.emit('Idle.')

    def primeFactors(self, n):
        i = 2
        factors = []
        while i * i <= n:
            if n % i:
                i += 1
            else:
                n //= i
                factors.append(i)
        if n > 1:
            factors.append(n)
        return factors


class Window(QtGui.QWidget):

    def __init__(self):
        QtGui.QWidget.__init__(self)
        self.button_start = QtGui.QPushButton('Start', self)
        self.button_cancel = QtGui.QPushButton('Cancel', self)
        self.label_status = QtGui.QLabel('', self)

        layout = QtGui.QVBoxLayout(self)
        layout.addWidget(self.button_start)
        layout.addWidget(self.button_cancel)
        layout.addWidget(self.label_status)

        self.setFixedSize(400, 200)

    @QtCore.pyqtSlot(str)
    def updateStatus(self, status):
        self.label_status.setText(status)


if __name__=='__main__':
    app = QtGui.QApplication(sys.argv)
    example = Example(app)
    sys.exit(app.exec_())

解决方案 2：

您说得对，让工作线程进行处理，而主线程进行 GUI 操作，这是一件好事。此外，PyQt 还提供了线程安全的信号/槽机制的线程检测。

这听起来很有趣。在他们的示例中，他们构建了一个 GUI

import sys, time
from PyQt4 import QtCore, QtGui

class MyApp(QtGui.QWidget):
 def __init__(self, parent=None):
  QtGui.QWidget.__init__(self, parent)

  self.setGeometry(300, 300, 280, 600)
  self.setWindowTitle('threads')

  self.layout = QtGui.QVBoxLayout(self)

  self.testButton = QtGui.QPushButton("test")
  self.connect(self.testButton, QtCore.SIGNAL("released()"), self.test)
  self.listwidget = QtGui.QListWidget(self)

  self.layout.addWidget(self.testButton)
  self.layout.addWidget(self.listwidget)

 def add(self, text):
  """ Add item to list widget """
  print "Add: " + text
  self.listwidget.addItem(text)
  self.listwidget.sortItems()

 def addBatch(self,text="test",iters=6,delay=0.3):
  """ Add several items to list widget """
  for i in range(iters):
   time.sleep(delay) # artificial time delay
   self.add(text+" "+str(i))

 def test(self):
  self.listwidget.clear()
  # adding entries just from main application: locks ui
  self.addBatch("_non_thread",iters=6,delay=0.3)

（简单的用户界面包含一个列表小部件，我们将通过单击按钮向其中添加一些项目）

然后你可以创建我们自己的线程类，一个例子是

class WorkThread(QtCore.QThread):
 def __init__(self):
  QtCore.QThread.__init__(self)

 def __del__(self):
  self.wait()

 def run(self):
  for i in range(6):
   time.sleep(0.3) # artificial time delay
   self.emit( QtCore.SIGNAL('update(QString)'), "from work thread " + str(i) )

  self.terminate()

您确实重新定义了run()方法。您可能会找到替代方法terminate()，请参阅教程。

解决方案 3：

在我看来，迄今为止最好的解释，其中的示例代码最初没有响应，然后得到改进，可以在这里找到。

请注意，这确实使用了所需的（非子类）QThread和moveToThread方法，文章声称这是首选方法。

上面链接的页面还提供了 PyQt5 等效于 C Qt 页面，该页面由 Maya Posch在 2011 年给出了权威解释。我认为她当时可能正在使用 Qt4，但该页面仍然适用于 Qt5（因此是 PyQt5）并且值得深入研究，包括许多评论（和她的回复）。

为了防止上面第一个链接有一天出现 404 错误（这太糟糕了！），这是基本 Python 代码，它相当于 Maya 的 C 代码：

self.thread = QtCore.QThread()
# Step 3: Create a worker object
self.worker = Worker()
# Step 4: Move worker to the thread
self.worker.moveToThread(self.thread)
# Step 5: Connect signals and slots
self.thread.started.connect(self.worker.run)
self.worker.finished.connect(self.thread.quit)
self.worker.finished.connect(self.worker.deleteLater)
self.thread.finished.connect(self.thread.deleteLater)
self.worker.progress.connect(self.reportProgress)
# Step 6: Start the thread
self.thread.start()

# Final resets
self.longRunningBtn.setEnabled(False)
self.thread.finished.connect(
    lambda: self.longRunningBtn.setEnabled(True)
)
self.thread.finished.connect(
    lambda: self.stepLabel.setText("Long-Running Step: 0")
)   

self注意：该页面示例中的NB是QMainWindow对象。我认为您可能必须非常小心将QThread实例附加到属性上：超出范围时会被销毁的实例，但具有QThread属性的实例，或者QThread超出范围的本地实例，似乎能够导致一些无法解释的 Python 崩溃，而这些崩溃不会被sys.excepthook(或sys.unraisablehook) 发现。建议谨慎使用。

...Worker看起来像这样：

class Worker(QtCore.QObject):
    finished = QtCore.pyqtSignal()
    progress = QtCore.pyqtSignal(int)

    def run(self):
        """Long-running task."""
        for i in range(5):
            sleep(1)
            self.progress.emit(i + 1)
        self.finished.emit()