多线程任务管理器

该代码实现了一个多线程工作者类（Worker）和信号类（WorkerSignals），用于在PyQt6应用程序中处理多线程任务。Worker类继承自QRunnable，WorkerSignals类继承自QObject，定义了一组可用的信号，以便在工作线程中传递信息。

代码概述

该代码实现了一个多线程工作者类（Worker）和信号类（WorkerSignals），用于在PyQt6应用程序中处理多线程任务。Worker类继承自QRunnable，WorkerSignals类继承自QObject，定义了一组可用的信号，以便在工作线程中传递信息。

依赖库

  • sys：用于获取异常信息。
  • traceback：用于格式化异常跟踪信息。
  • PyQt6.QtCore：提供PyQt6核心功能，包括信号和槽机制、可运行对象等。

类和方法

WorkerSignals 类

WorkerSignals 继承自 QObject，定义了一组可用的信号，以便在工作线程中传递信息。

信号：

  • finished：没有数据，表示任务完成。
  • error：传递一个包含异常类型、异常值和格式化异常跟踪信息的元组。
  • result：传递处理函数返回的结果数据。
  • progress：传递一个整数，表示进度百分比。
  • update_ui：用于更新用户界面。
  • tableUpdate：用于更新表格。
  • addItem：传递一个整数和一个QVariant，用于添加项目。
  • deleteItem：传递一个整数，用于删除项目。
  • taskStart：表示任务开始。
  • taskDone：表示任务完成。
class WorkerSignals(QObject):
    '''
    Defines the signals available from a running worker thread.
    Supported signals are:

    finished
        No data

    error
        `tuple` (exctype, value, traceback.format_exc() )

    result
        `object` data returned from processing, anything

    progress
        `int` indicating % progress

    '''
    finished = pyqtSignal()
    error = pyqtSignal(tuple)
    result = pyqtSignal(object)
    progress = pyqtSignal(int)
    update_ui = pyqtSignal()
    tableUpdate = pyqtSignal()
    addItem = pyqtSignal(int, QtCore.QVariant)
    deleteItem = pyqtSignal(int)
    taskStart = pyqtSignal()
    taskDone = pyqtSignal()

Worker 类

Worker 继承自 QRunnable，实现了一个工作线程，用于处理耗时的任务。

方法：

  • __init__(self, fn, *args, **kwargs)：初始化工作线程。

    • fn：要在线程中运行的函数。
    • args：传递给函数的参数。
    • kwargs：传递给函数的关键字参数。
    • 初始化时存储函数和参数，并创建一个WorkerSignals实例。

  • run(self)：运行工作线程。

    • 尝试运行传入的函数，并传递参数和关键字参数。
    • 如果出现异常，捕获并发出error信号。
    • 最后，无论是否有异常，都会发出finished信号，表示任务完成。
class Worker(QRunnable):
    '''
    Worker thread
    Inherits from QRunnable to handle worker thread setup, signals and wrap-up.

    :param callback: The function callback to run on this worker thread. Supplied args and
                     kwargs will be passed through to the runner.
    :type callback: function
    :param args: Arguments to pass to the callback function
    :param kwargs: Keywords to pass to the callback function

    '''

    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()

        # Store constructor arguments (re-used for processing)
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()

        # Add the callback to our kwargs
        self.kwargs['progress_callback'] = self.signals.progress

    @pyqtSlot()
    def run(self):
        """
        Initialise the runner function with passed args, kwargs.
        """

        # Retrieve args/kwargs here; and fire processing using them
        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()))
        finally:
            self.signals.finished.emit()  # Done

总结

该代码定义了一个用于多线程任务处理的工作者类（Worker）和信号类（WorkerSignals）。Worker类继承自QRunnable，可以在工作线程中运行传入的函数，并通过信号机制传递任务的进度、结果或错误信息。WorkerSignals类定义了一组信号，以便在工作线程和主线程之间进行通信。