来自教程我看到信号和插槽可以用来从worker线程回调到主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能把响应信号发回给工作线程。我也不清楚为什么连接maingui的signal mainthread_callback_to_worker 导致崩溃（见注释的那行）。
我知道，一个变通的办法是让 do_something 来返回一些值，然后在worker中使用它作为 "确认"。但如果可能的话，我想知道使用信号&插槽的解决方案。
要了解错误的原因，必须在终端中运行代码，你会得到以下错误信息。
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只是一个活在主线程中的接口，只有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)