python - 使用 QRunnable 进行线程处理 - 发送双向回调的正确方式

Question

From a tutorial ，我已经看到信号和槽可用于从工作线程回调到主 GUI 线程，但我不确定如何使用信号和槽建立双向通信。以下是我正在使用的内容:

class RespondedToWorkerSignals(QObject):
    callback_from_worker = pyqtSignal()

class RespondedToWorker(QRunnable):
    def __init__(self, func, *args, **kwargs):
        super(RespondedToWorker, self).__init__()
        self._func = func
        self.args = args
        self.kwargs = kwargs
        self.signals = RespondedToWorkerSignals()
        self.kwargs['signal'] = self.signals.callback_from_worker
        print("Created a responded-to worker")

    @pyqtSlot()
    def run(self):
        self._func(*self.args, **self.kwargs)

    @pyqtSlot()
    def acknowledge_callback_in_worker(self):
        print("Acknowledged Callback in Worker")

class MainWindow(QMainWindow):

    # Signal meant to connect to a slot present within a worker
    mainthread_callback_to_worker = pyqtSignal()

    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)
        # Quick UI setup
        w, lay = QWidget(), QVBoxLayout()
        w.setLayout(lay)
        self.setCentralWidget(w)
        self.timer_label = QLabel("Timer Label")
        lay.addWidget(self.timer_label)
        self.btn_thread_example = QPushButton("Push Me")
        self.btn_thread_example.pressed.connect(self.thread_example)
        lay.addWidget(self.btn_thread_example)
        self.threadpool = QThreadPool()
        self.show()

        # Set up QTimer to continue in the background to help demonstrate threading advantage
        self.counter = 0
        self.timer = QTimer()
        self.timer.setInterval(1000)
        self.timer.timeout.connect(self.recurring_timer)
        self.timer.start()

    @pyqtSlot()
    def do_something(self, signal):
        # signal argument will be the callback_from_worker and it will emit to acknowledge_callback_in_mainthread
        print("do_something is sleeping briefly. Try to see if you get a locked widget...")
        time.sleep(7)
        signal.emit()

    @pyqtSlot()
    def acknowledge_callback_in_mainthread_and_respond(self):
        # this function should respond to callback_from_worker and emit a response
        print("Acknowledged Callback in Main")
        self.mainthread_callback_to_worker.emit()

    def thread_example(self):
        print("Beginning thread example")
        worker = RespondedToWorker(self.do_something)
        worker.signals.callback_from_worker.connect(self.acknowledge_callback_in_mainthread_and_respond)
    # self.mainthread_callback_to_worker.connect(worker.acknowledge_callback_in_worker) # <-- causes crash

    def recurring_timer(self):
        self.counter += 1
        self.timer_label.setText(f"Counter: {self.counter}")

if __name__ == '__main__':
    app = QApplication(sys.argv)
    win = MainWindow()
    app.setStyle("Fusion")
    win.show()
    sys.exit(app.exec())

目前，脚本可以创建第二个线程并向主 GUI 发送信号。我希望 GUI 将响应信号发送回工作线程。我也不确定为什么要连接主/GUI的信号 mainthread_callback_to_worker导致崩溃(请参阅注释掉的行)。
我了解一种解决方法是针对 do_something返回一些值，然后在工作人员内部将其用作“确认”。但如果可能的话，我想知道使用信号和插槽的解决方案。

Answer 1

要了解错误的原因，您必须在终端中运行代码，您将收到以下错误消息:

QObject::connect: Cannot connect MainWindow::mainthread_callback_to_worker() to (nullptr)::acknowledge_callback_in_worker()
Traceback (most recent call last):
  File "main.py", line 72, in thread_example
    self.mainthread_callback_to_worker.connect(worker.acknowledge_callback_in_worker) # <-- causes crash
TypeError: connect() failed between MainWindow.mainthread_callback_to_worker[] and acknowledge_callback_in_worker()
Aborted (core dumped)

错误的原因是滥用 pyqtSlot装饰器，因为它只能用于 QObject方法，但 QRunnable 不会导致该异常，此外，在非 QObject 中它没有利用任何优势，因此 run() 中的装饰器方法没有意义。

另一方面，QRunnable 只是一个存在于主线程中的接口(interface)，并且只有 run 方法在另一个线程中执行，因此 QRunnable 不能成为 worker ，因为该类型的目标必须在辅助线程中执行其方法。

因此，使用上述 QRunnable 不是合适的选项，因此出于您的目的，我建议使用位于辅助线程中并调用方法的 QObject。

import sys
import time

from PyQt5.QtCore import pyqtSignal, pyqtSlot, QObject, QTimer, QThread
from PyQt5.QtWidgets import (
    QApplication,
    QMainWindow,
    QWidget,
    QVBoxLayout,
    QLabel,
    QPushButton,
)


class Worker(QObject):
    callback_from_worker = pyqtSignal()

    def __init__(self, func, *args, **kwargs):
        super(Worker, self).__init__()
        self._func = func
        self.args = args
        self.kwargs = kwargs
        self.kwargs["signal"] = self.callback_from_worker

    def start_task(self):
        QTimer.singleShot(0, self.task)

    @pyqtSlot()
    def task(self):
        self._func(*self.args, **self.kwargs)

    @pyqtSlot()
    def acknowledge_callback_in_worker(self):
        print("Acknowledged Callback in Worker")
        print(threading.current_thread())


class MainWindow(QMainWindow):
    mainthread_callback_to_worker = pyqtSignal()

    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)

        w, lay = QWidget(), QVBoxLayout()
        w.setLayout(lay)
        self.setCentralWidget(w)
        self.timer_label = QLabel("Timer Label")
        lay.addWidget(self.timer_label)
        self.btn_thread_example = QPushButton("Push Me")
        self.btn_thread_example.pressed.connect(self.thread_example)
        lay.addWidget(self.btn_thread_example)

        self.counter = 0
        self.timer = QTimer(interval=1000, timeout=self.recurring_timer)
        self.timer.start()

        self._worker = Worker(self.do_something)
        self._worker.callback_from_worker.connect(
            self.acknowledge_callback_in_mainthread_and_respond
        )

        self.worker_thread = QThread(self)
        self.worker_thread.start()
        self._worker.moveToThread(self.worker_thread)

    @pyqtSlot()
    def do_something(self, signal):
        print(
            "do_something is sleeping briefly. Try to see if you get a locked widget..."
        )
        time.sleep(7)
        signal.emit()

    @pyqtSlot()
    def acknowledge_callback_in_mainthread_and_respond(self):
        print("Acknowledged Callback in Main")
        self.mainthread_callback_to_worker.emit()

    def thread_example(self):
        print("Beginning thread example")
        self._worker.start_task()

    def recurring_timer(self):
        self.counter += 1
        self.timer_label.setText(f"Counter: {self.counter}")


if __name__ == "__main__":
    app = QApplication(sys.argv)
    win = MainWindow()
    app.setStyle("Fusion")
    win.show()
    ret = app.exec_()
    win.worker_thread.quit()
    win.worker_thread.wait()
    sys.exit(ret)