PyQt5 GUI防冻结：如何向QObject Worker类传入自定义函数

现有代码的问题 #

• 你写的Worker构造方法没有定义target、args、kwargs参数，也没有把这些参数存为实例属性，run方法执行时会直接抛出AttributeError
• 移除了原有的progress进度信号，自定义任务需要上报进度时没有对应的触发通道

修正后的Worker类 #

from PyQt5.QtCore import QObject, pyqtSignal

class Worker(QObject):
    # 支持返回任意类型的执行结果
    finished = pyqtSignal(object)
    # 通用进度信号，可根据实际需求调整参数类型
    progress = pyqtSignal(int)

    def __init__(self, target, *args, **kwargs):
        super().__init__()
        # 存储待执行的自定义目标函数
        self._target = target
        # 存储目标函数需要的位置参数
        self._args = args
        # 存储目标函数需要的关键字参数
        self._kwargs = kwargs

    def run(self):
        try:
            # 执行传入的自定义函数
            result = self._target(*self._args, **self._kwargs)
            # 执行完成后发射结果信号
            self.finished.emit(result)
        except Exception as e:
            # 直接把异常作为finished信号的发射内容，避免子线程崩溃无反馈
            self.finished.emit(e)

调用方式说明 #

原有窗口类的线程创建、信号绑定逻辑不需要做大的改动，只需要在实例化Worker时直接传入自定义函数、对应参数即可：

# 自定义长任务示例，支持接收进度回调上报执行进度
def my_custom_task(progress_cb=None):
    for i in range(5):
        sleep(1)
        # 如果需要上报进度，调用传入的回调函数即可
        if progress_cb:
            progress_cb(i + 1)
    return "长任务执行完成"

def runLongTask(self):
    # QThread创建逻辑和原有代码完全一致
    self.thread = QThread()
    # 实例化Worker时直接传入自定义函数、参数
    self.worker = Worker(
        target=self.my_custom_task,
        progress_cb=self.worker.progress.emit
    )
    # 后续线程移动、信号绑定逻辑和原有代码一致
    self.worker.moveToThread(self.thread)
    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)
    # 新增绑定任务结果的处理逻辑
    self.worker.finished.connect(self.handle_task_result)
    # 启动线程
    self.thread.start()

    # 原有按钮禁用、状态重置逻辑不变
    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"))

def handle_task_result(self, result):
    # 处理任务返回结果，优先判断是否是异常
    if isinstance(result, Exception):
        self.stepLabel.setText(f"任务执行出错: {str(result)}")
    else:
        self.stepLabel.setText(f"任务返回: {result}")

注意事项 #

• 禁止在子线程的自定义函数中直接操作GUI组件，所有界面更新必须通过信号槽跨线程通信实现，否则会出现不可预期的程序崩溃
• 如果长任务不需要上报进度，不用传入progress_cb参数即可，Worker会自动跳过进度发射逻辑
• 可根据需求额外新增error信号，单独处理子线程的异常，避免把异常和普通结果混为一谈

内容的提问来源于stack exchange，提问作者John Herwig