multithreading - 在 QThread.exit() 上立即停止处理事件队列

Question

我正在构建一个 Qt GUI 应用程序，它使用 QThread/QObject 组合充当在主线程之外执行操作的工作程序。

通过moveToThread，QObject 被移动到QThread 中。这样，我的工作人员就可以拥有在事件循环(由 QThread 提供)中处理的信号(它是 QObject)和槽。

现在我想让 worker 以一种特殊的方式运行，只要事件循环中的槽遇到 Python 异常，它们就会优雅地停止线程。

通过一些测试，我发现在 PyQt5 中，槽中的一个异常导致整个应用程序停止，据我所知，与 PyQt4 相比，这是一个有意的改变，在 PyQt4 中只打印了异常，但事件循环保持不变运行。我读到可以通过将您自己的“excepthook”猴子修补到 sys.excepthook 来避免这种情况，Qt 以停止解释器的方式实现它。

所以我做到了，到目前为止这很有效。此外，excepthook 使我能够在发生异常时 exit() 我的工作人员，对此我在其他地方找不到更好的方法。我尝试子类化 QThread 并在 QThread 的 run() 方法中围绕对 exec_() 的调用放置一个 try..except，但它没有'传播事件循环中发生的异常...所以剩下的唯一选择是将 try..except block 放在每个插槽中，我想避免这种情况。还是我错过了什么？

下面是一个 MWE，它展示了我目前所拥有的。我的问题是当异常发生时退出线程不会立即发生，用 error 槽演示，这导致调用 thread.exit()除了钩子(Hook)。相反，线程事件循环中的所有其他剩余事件都将被执行，这里由我安排在它后面的 do_work 插槽演示。 exit() 似乎只是将另一个事件安排到队列中，一旦它被处理，事件循环就会停止。

我该如何解决这个问题？有没有办法刷新 QThread 的事件队列？我能以某种方式优先退出吗？

或者也许是另一种完全不同的方式来捕获槽中的异常并使线程停止，而不停止主程序？

代码:

import sys
import time
from qtpy import QtWidgets, QtCore


class ThreadedWorkerBase(QtCore.QObject):
    def __init__(self):
        super().__init__()
        self.thread = QtCore.QThread(self)
        self.thread.setTerminationEnabled(False)
        self.moveToThread(self.thread)
        self.thread.start()

    def schedule(self, slot, delay=0):
        """ Shortcut to QTimer's singleShot. delay is in seconds. """
        QtCore.QTimer.singleShot(int(delay * 1000), slot)


class Worker(ThreadedWorkerBase):
    test_signal = QtCore.Signal(str)   # just for demo

    def do_work(self):
        print("starting to work")
        for i in range(10):
            print("working:", i)
            time.sleep(0.2)

    def error(self):
        print("Throwing error")
        raise Exception("This is an Exception which should stop the worker thread's event loop.")


#  set excepthook to explicitly exit Worker thread after Exception
sys._excepthook = sys.excepthook
def excepthook(type, value, traceback):
    sys._excepthook(type, value, traceback)
    thread = QtCore.QThread.currentThread()
    if isinstance(thread.parent(), ThreadedWorkerBase):
        print("This is a Worker thread. Exiting...")
        thread.exit()
sys.excepthook = excepthook

# create demo app which schedules some tasks
app = QtWidgets.QApplication([])
worker = Worker()
worker.schedule(worker.do_work)
worker.schedule(worker.error)    # this should exit the thread => no more scheduling
worker.schedule(worker.do_work)
worker.thread.wait()   # worker should exit, just wait...

输出:

starting to work
working: 0
working: 1
working: 2
working: 3
working: 4
working: 5
working: 6
working: 7
working: 8
working: 9
Throwing error
Traceback (most recent call last):
  File "qt_test_so.py", line 31, in error
    raise Exception("This is an Exception which should stop the worker thread's event loop.")
Exception: This is an Exception which should stop the worker thread's event loop.
This is a Worker thread. Exiting...
starting to work
working: 0
working: 1
working: 2
working: 3
working: 4
working: 5
working: 6
working: 7
working: 8
working: 9

期望:

输出应该在“Exiting...”之后结束。

Answer 1

QThread.exit 的 Qt 文档有点误导:

Tells the thread's event loop to exit with a return code.

After calling this function, the thread leaves the event loop and returns from the call to QEventLoop::exec(). The QEventLoop::exec() function returns returnCode.

By convention, a returnCode of 0 means success, any non-zero value indicates an error.

Note that unlike the C library function of the same name, this function does return to the caller -- it is event processing that stops. [emphasis added]

这表明在调用 exit() 之后，不会再对线程的事件队列进行进一步处理。但事实并非如此，因为 QEventLoop 总是调用 processEvents before 检查它是否应该退出。这意味着当 exec() 返回时，事件队列将始终为空。

在您的示例中，单次计时器会将事件发布到接收线程的事件队列，最终将调用连接的插槽。因此，无论您做什么，所有这些插槽都将在线程最终退出之前被调用。

解决这个问题的一个相当简单的方法是使用 requestInterruption具有装饰器的功能，用于检查是否应调用插槽:

def interruptable(slot):
    def wrapper(self, *args, **kwargs):
        if not self.thread.isInterruptionRequested():
            slot(self, *args, **kwargs)
    return wrapper

class Worker(ThreadedWorkerBase):
    test_signal = QtCore.pyqtSignal(str)   # just for demo

    @interruptable
    def do_work(self):
        print("starting to work")
        for i in range(10):
            print("working:", i)
            time.sleep(0.2)

    @interruptable
    def error(self):
        print("Throwing error")
        raise Exception("This is an Exception which should stop the worker thread's event loop.")

def excepthook(type, value, traceback):
    sys.__excepthook__(type, value, traceback)
    thread = QtCore.QThread.currentThread()
    if isinstance(thread.parent(), ThreadedWorkerBase):
        print("This is a Worker thread. Exiting...")
        thread.requestInterruption()
        thread.exit()
sys.excepthook = excepthook