使用 QRunnable 进行线程化 - 发送双向回调的正确方式
Threading with QRunnable - Proper manner of sending bi-directional callbacks
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 将响应信号发送回工作线程。我也不确定为什么连接 main/GUI 的信号 mainthread_callback_to_worker 会导致崩溃（请参阅注释掉的行）。
我知道一种解决方法是 do_something 到 return 一些值，然后在工作人员内部使用它作为“确认”。但如果可能的话，我想知道使用信号和插槽的解决方案。
要了解错误原因，您必须 运行 终端中的代码，您将收到以下错误消息：
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 只是一个存在于主线程中的接口，并且只有 运行 方法在另一个线程中执行，因此 QRunnable 不能成为工作者，因为 objective 必须在辅助线程中执行其方法。
因此上面的 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)
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 将响应信号发送回工作线程。我也不确定为什么连接 main/GUI 的信号 mainthread_callback_to_worker 会导致崩溃（请参阅注释掉的行）。
我知道一种解决方法是 do_something 到 return 一些值，然后在工作人员内部使用它作为“确认”。但如果可能的话，我想知道使用信号和插槽的解决方案。
要了解错误原因，您必须 运行 终端中的代码，您将收到以下错误消息：
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 只是一个存在于主线程中的接口，并且只有 运行 方法在另一个线程中执行，因此 QRunnable 不能成为工作者，因为 objective 必须在辅助线程中执行其方法。
因此上面的 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)