PyQt5如何多个QRunnable报告进度和最终结果

Posted 2023-03-30

【中文标题】PyQt5如何多个QRunnable报告进度和最终结果

【英文标题】：PyQt5 how multiple QRunnable report progress and final result

【发布时间】：2020-10-22 02:01:07

【问题描述】：

我的代码看起来像这个 *** link 的答案。 代码如下：

from PyQt5 import QtCore, QtGui, QtWidgets

import time
import traceback, sys

uart_result = ['1','2', '3', '4', '5', '6']

#Running all these methods in parallel
@QtCore.pyqtSlot()
def run1():
    print("Job 1")
    return uart_result

@QtCore.pyqtSlot()
def run2():
    print("Job 2")
    return uart_result

@QtCore.pyqtSlot()
def run3():
    print("Job 3")
    return uart_result

@QtCore.pyqtSlot()
def run4():
    print("Job 4")
    return uart_result

class WorkerSignals(QtCore.QObject):
    finished = QtCore.pyqtSignal()
    error = QtCore.pyqtSignal(tuple)
    result = QtCore.pyqtSignal(object)
    progress = QtCore.pyqtSignal(int)


class Worker(QtCore.QRunnable):
    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()
    def run(self):
        try:
            result = self.fn(*self.args, **self.kwargs)
        except:
            traceback.print_exc()
            exctype, value = sys.exc_info()[:2]
            self.signals.error.emit((exctype, value, traceback.format_exc()))
        else:
            self.signals.result.emit(result)  # Return the result of the processing
        finally:
            self.signals.finished.emit()  # Done



class MainWindow(QtWidgets.QMainWindow):
    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)
        b = QtWidgets.QPushButton("START!")
        b.pressed.connect(self.runner)

        w = QtWidgets.QWidget()
        layout = QtWidgets.QVBoxLayout(w)
        layout.addWidget(b)
        self.setCentralWidget(w)
        auto_update_worker = Worker(self.detect_auto_update_status)#edit
        QtCore.QThreadPool.globalInstance().start(auto_update_worker)#edit
         
    # edit
    def detect_auto_update_status(self):
        while True:
            print('get_auto_update_data_from_server')
            time.sleep(2)

    def print_output(self, uart_list):
        print(uart_list)

    def thread_complete(self):
        print("THREAD COMPLETE!")

    def runner(self):
        thread_pool = QtCore.QThreadPool.globalInstance()
        print("Multithreading with maximum %d threads" % thread_pool.maxThreadCount())
        print("You pressed the Test button")
        for task in (run1, run2, run3, run4):
            worker = Worker(task)
            worker.signals.result.connect(self.print_output)
            worker.signals.finished.connect(self.thread_complete)
            thread_pool.start(worker)

if __name__ == '__main__':
    import sys
    app = QtWidgets.QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec_())

我可以从下面的函数中得到每个任务（QRunnale）的结果

    def print_output(self, uart_list):
        print(uart_list)

并在函数下方获取完成状态表单

    def thread_complete(self):
        print("THREAD COMPLETE!")

我的问题是，我如何知道所有任务何时完成（为下一​​步做其他事情）以及如何在任务运行期间报告进度？

【参考方案1】：

一个可能的解决方案是在每个 QRunnable 完成运行后检查活动线程（activeThreadCount()）：

def thread_complete(self):
    print("THREAD COMPLETE!")
    thread_pool = QtCore.QThreadPool.globalInstance()
    if thread_pool.activeThreadCount() == 0:
        print("finished")

【讨论】：

在一个应用程序中，我可能让另一个线程一次又一次地检测 auto_update 数据或其他数据。在这种情况下，thread_pool.activeThreadCount() == 0 似乎无法正常工作。

@TimYu 然后提供一个真实的minimal reproducible example

我已经编辑了代码，添加了一个新的工作人员来检测来自服务器的 auto_update 数据。

@TimYu 我不明白你在用那个版本尝试什么，你能详细解释一下吗

我想知道这些任务（run1,run2,run3,run4）什么时候完成。因为 QtCore.QThreadPool.globalInstance() 是一个全局实例并且 auto_update 线程始终处于活动状态，在这种情况下，我们不能使用 'thread_pool.activeThreadCount() == 0' 来检测这些任务（run1,run2,run3,run4 ) 完成与否