PyQt5:从工作线程安全地调用小部件插槽

Question

如何从QThread worker调用小部件插槽？我知道我可以为每个小部件的插槽创建一个信号，如下所示：

class App(QtWidgets.QMainWindow):
signal_line_edit_1_setText = pyqtSignal(str)
signal_line_edit_2_setText = pyqtSignal(str)
...
def __init__(self):
...
self.signal_line_edit_1_setText.connect(self.line_edit_1.setText)
self.signal_line_edit_2_setText.connect(self.line_edit_2.setText)
self.worker = Worker(self)

class Worker(QThread):
#  Maybe I have to create signals for Worker class and then connect them to app's signals,
#  but that would be even more complicated
def __init__(self, app):
self.app = app
...
def run(self):
self.app.signal_line_edit_1_setText.emit('Worker Running')
...

难道没有一种更简单的方法可以让线程安全地与小部件交互吗？QTimer在没有信号包装的情况下工作，但它使UI有点滞后。我知道QThreadPool，但并不真正了解它。

Answer 1

有一种方法可以完全按照您的意愿进行操作，但它的使用频率远低于自定义信号(无论如何，在Python中，我对C++不太确定(。

相关的API是QMetaObject.invokeMethod。这允许对QObject子类的任何方法进行线程安全调用，只要该方法可以通过Qt元对象系统访问。在实践中，这通常会将其限制为预先存在的Qt方法，以及用pyqtSlot装饰器包装的用户定义方法。以下是典型用法：

QtCore.QMetaObject.invokeMethod(widget, 'mySlot', QtCore.Q_ARG(int, number))

正如你所看到的，与PyQt的新型信号和槽语法相比，它看起来相当笨拙，事实上，它的缺点与老式信号和槽句法相似：即，它有点容易出错，冗长，而且不太像蟒蛇。它唯一的显著优点(相对于当前用例(是避免了必须预先定义信号。(请参阅下面的演示脚本，该脚本使用这两种方法来确保在主线程中执行GUI更新(。

我想可以创建一个自定义类，使用这种方法(可能通过__getattr__(跨线程自动调用方法。但是，既然已经有了实现相同功能的内置机制，为什么还要麻烦开发和维护这样一个类呢？定义一个自定义信号，将其连接到一个可调用信号，并发出它一点也不复杂：

class Worker(QThread):
customSignal = pyqtSignal(int)
def run(self):
self.customSignal.emit(42)
worker = Worker()
worker.customSignal.connect(lambda x: print(x))
worker.start()

生成的代码可读性强、灵活且易于维护。

---

演示脚本：

from PyQt5 import QtCore, QtWidgets
def thread_id():
return int(QtCore.QThread.currentThreadId())
class Worker(QtCore.QThread):
progressChanged = QtCore.pyqtSignal(int)
def setMethod(self, invoke=False):
self._invoke = invoke
def run(self):
print()
print(f'Thread: {MAIN_THREAD} [Main]')
print(f'Thread: {thread_id()} [Worker.run]')
invoke = getattr(self, '_invoke', False)
print('Using Method:', 'invoke' if invoke else 'signal')
for count in range(1, 6):
self.msleep(500)
if invoke:
QtCore.QMetaObject.invokeMethod(
self.parent(), 'updateProgress', QtCore.Q_ARG(int, count))
else:
self.progressChanged.emit(count)
class Window(QtWidgets.QWidget):
def __init__(self):
super().__init__()
self.button = QtWidgets.QPushButton('Test')
self.button.clicked.connect(self.handleButton)
self.check = QtWidgets.QCheckBox('Use inkoke')
self.label = QtWidgets.QLabel()
self.label.setAlignment(QtCore.Qt.AlignCenter)
layout = QtWidgets.QGridLayout(self)
layout.addWidget(self.label, 0, 0, 1, 2)
layout.addWidget(self.button, 1, 0)
layout.addWidget(self.check, 1, 1)
self.worker = Worker(self)
self.worker.progressChanged.connect(self.updateProgress)
self.updateProgress()
def handleButton(self):
if not self.worker.isRunning():
self.updateProgress()
self.worker.setMethod(invoke=self.check.isChecked())
self.worker.start()
@QtCore.pyqtSlot(int)
def updateProgress(self, count=0):
if count:
print(f'Thread: {MAIN_THREAD} [Main]')
print(f'Thread: {thread_id()} [Window.updateProgress]')
self.label.setText(f'Count: {count}')
def closeEvent(self, event):
self.worker.quit()
self.worker.wait()

app = QtWidgets.QApplication(['Test'])
MAIN_THREAD = thread_id()
print(f'Thread: {MAIN_THREAD} [Main]')
window = Window()
window.setGeometry(600, 100, 300, 200)
window.show()
app.exec_()