将函数传递给Python GUI应用程序中的Worker(QObject)类，以防止冻结/阻塞

Question

我正试图找到一种方法，使用PyQT5成功地将函数传递给Python中的Worker类。我希望能够将自定义函数传递给Worker类，而不是在示例Worker类代码中使用预定义的运行函数(或长期运行任务(。下面我粘贴了我正在使用的示例代码，然后进行了我尝试过的调整。

from time import sleep
from PyQt5.QtCore import QObject, QThread, pyqtSignal,Qt
from PyQt5.QtWidgets import QApplication,QMainWindow,QLabel,QPushButton,QVBoxLayout,QWidget
import sys
# Step 1: Create a worker class
class Worker(QObject):
finished = pyqtSignal()
progress = pyqtSignal(int)

def run(self):
"""Long-running task."""
for i in range(5):
sleep(1)
self.progress.emit(i + 1)
self.finished.emit()
class Window(QMainWindow):
def __init__(self, parent=None):
super().__init__(parent)
self.clicksCount = 0
self.setupUi()
def setupUi(self):
self.setWindowTitle("Freezing GUI")
self.resize(300, 150)
self.centralWidget = QWidget()
self.setCentralWidget(self.centralWidget)
# Create and connect widgets
self.clicksLabel = QLabel("Counting: 0 clicks", self)
self.clicksLabel.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
self.stepLabel = QLabel("Long-Running Step: 0")
self.stepLabel.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
self.countBtn = QPushButton("Click me!", self)
self.countBtn.clicked.connect(self.countClicks)
self.longRunningBtn = QPushButton("Long-Running Task!", self)
self.longRunningBtn.clicked.connect(self.runLongTask)
# Set the layout
layout = QVBoxLayout()
layout.addWidget(self.clicksLabel)
layout.addWidget(self.countBtn)
layout.addStretch()
layout.addWidget(self.stepLabel)
layout.addWidget(self.longRunningBtn)
self.centralWidget.setLayout(layout)
def countClicks(self):
self.clicksCount += 1
self.clicksLabel.setText(f"Counting: {self.clicksCount} clicks")
def reportProgress(self, n):
self.stepLabel.setText(f"Long-Running Step: {n}")
def runLongTask(self):
# Step 2: Create a QThread object
self.thread = QThread()
# Step 3: Create a worker object
self.worker = Worker()
# Step 4: Move worker to the thread
self.worker.moveToThread(self.thread)
# Step 5: Connect signals and slots
self.thread.started.connect(self.worker.run)
self.worker.finished.connect(self.thread.quit)
self.worker.finished.connect(self.worker.deleteLater)
self.thread.finished.connect(self.thread.deleteLater)
self.worker.progress.connect(self.reportProgress)
# Step 6: Start the thread
self.thread.start()
# Final resets
self.longRunningBtn.setEnabled(False)
self.thread.finished.connect(
lambda: self.longRunningBtn.setEnabled(True)
)
self.thread.finished.connect(
lambda: self.stepLabel.setText("Long-Running Step: 0")
)
app = QApplication(sys.argv)
win = Window()
win.show()
sys.exit(app.exec())

---

class Worker(QObject):
finished = pyqtSignal(str)

def __init__(self, *init_args, **init_kwargs):
QObject.__init__(self, *init_args, **init_kwargs)
self._return = None
def run(self):
"""Long-running task."""
self._return = self._target(*self._args, **self._kwargs)
self.finished.emit(self._return)

Answer 1

作为解决方案之一，您可以将函数和参数传递给Worker的__init__方法，如：

class Worker(QObject):
finished = pyqtSignal()
progress = pyqtSignal(int)
result = pyqtSignal('QVariant')
def __init__(self, function, args):
super().__init__()
self.function = function
self.args = args
def run(self):
res = self.function(self.args)
self.result.emit(res)
self.finished.emit()

因此，您可以在创建工作者时传递一个方法和一个参数，并连接该方法来处理结果：

self.thread = QThread()
self.worker = Worker(Proxy.GetInfo, code)
self.worker.moveToThread(self.thread)
self.thread.started.connect(self.worker.run)
self.worker.finished.connect(self.thread.quit)
self.worker.finished.connect(self.thread.deleteLater)
self.worker.result.connect(self.receiveResult)
self.thread.finished.connect(self.thread.deleteLater)
self.thread.start()