如何构造大型pyqt5 GUI，而无需将QThread子类化并使用QPushButtons来执行长时间运行的任务

Question

我正在使用PyQt5 GUI创建程序。该程序将从带有许多按钮的UI开始。这些按钮将用于打开其他程序/已完成的长期运行任务。我知道我需要使用QThread，但是我不确定如何构造程序以便正确缩放。

我已经有很长时间了，已经阅读了许多帖子/教程。最倾向于子类化路线。过去，我设法创建了一个工作程序来继承QThread的子类，但是从那以后，我读到这种方法不是优选的。

我有一种感觉，我应该创建一个通用工作者，并传递带有* args和** kwargs的函数，但这还不属于我的技能范围。

我最初在GUI初始化期间为每个按钮创建了一个线程，但似乎很快就会失去控制。

我目前处于在与button.clicked信号相连的插槽下创建线程的阶段。我不确定每个按钮是否必须要有一个工作程序，或者是否可以/应该使一个通用工作程序并传递一个函数。注意：我已尝试执行此操作，但无法执行此操作。

#Import standard modules
import sys

#Import third-party modles
from PyQt5.QtCore import QObject, QThread, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QMainWindow, QApplication, QPushButton, QVBoxLayout, QWidget

class Worker(QObject):

    #Custom signals?? or built-in QThread signals?
    started = pyqtSignal()    
    finished = pyqtSignal()

    def __init__(self): 
        super().__init__()
        self.started.emit()

    @pyqtSlot()
    def do_something(self):
            for _ in range(3):
                print('Threading...')
                QThread.sleep(1)
            self.finished.emit()

class Window(QMainWindow):

    def __init__(self):
        super().__init__()
        self.initUi()

    def initUi(self):
        #Create GUI
        self.centralWidget  = QWidget()
        self.setCentralWidget(self.centralWidget )  
        self.vertical_layout = QVBoxLayout(self.centralWidget)
        self.setWindowTitle('QThread Test')
        self.setGeometry(300, 300, 300, 50)
        self.button1=QPushButton("Task 1", self, clicked=self._task1_clicked)
        self.button2=QPushButton("Task 2", self, clicked=self._task2_clicked)
        self.vertical_layout.addWidget(self.button1)
        self.vertical_layout.addWidget(self.button2)
        self.vertical_layout.addStretch()

    def _task1_clicked(self):
        print('task1 clicked')

        #Create the worker
        self.my_worker = Worker()          

        #Create thread; needs to be done before connecting signals/slots
        self.task1_thread = QThread()

        #Move the worker to the thread
        self.my_worker.moveToThread(self.task1_thread)        

        #Connect worker and thread signals to slots
        self.task1_thread.started.connect(self._thread_started)
        self.task1_thread.started.connect(self.my_worker.do_something)
        self.my_worker.finished.connect(self._thread_finished) 

        #Start thread
        self.task1_thread.start() 

    def _task2_clicked(self):
        print('task2 clicked')

    def _thread_started(self):
        print('thread started')

    def _thread_finished(self):
        print('thread finished')
        self.my_worker.isRunning = False
        self.task1_thread.quit()
        self.task1_thread.wait()
        print('The thread is running: ' + str(self.task1_thread.isRunning()))


if __name__ == '__main__':
    app = QApplication(sys.argv)
    form = Window()
    form.show()
    app.exec_()

上面的方法似乎有效，但是我觉得我偶然发现了它，但这不是正确的方法。如果它完全错误，我不希望这成为我的“首选”方法。与一个按钮/一个任务程序相比，我希望能够生成更复杂的（执行更多按钮的程序）。

另外，我似乎无法从根本上使QThread的启动和完成信号无法自定义生成。这是我认为我要解决此错误的原因之一。

Answer 1

from PyQt5 import QtCore

class AsyncTask(QtCore.QThread):
    taskDone = QtCore.pyqtSignal(dict)
    def __init__(self, *, task, callback=None, parent = None):
        super().__init__(parent)
        self.task = task
        if callback != None:
            self.taskDone.connect(callback)
        if callback == None:
            callback = self.callback
        self.start()

    def run(self):
        try:
            result = self.task()
            print(result)
            self.taskDone.emit(result)
        except Exception as ex:
            print(ex)

    def callback(self):
        print('callback')

请尝试上面的代码，像这样调用：AsyncTask（task = yourTaskFunction，callback = yourCallbackFunction）