如何基于PyQt5实现独立窗口进度条以展示耗时函数进度？

核心原理 #

PyQt的主线程负责UI渲染，耗时操作必须放在子线程中执行，否则会导致主窗口卡死。通过QThread创建子线程运行耗时函数，再利用信号槽机制将进度数据从子线程传递到主线程，更新进度条UI。

具体实现步骤 #

1. 创建子线程类：继承QThread，将耗时函数complicated_function放在线程的run()方法中，并定义进度更新信号。
2. 设计进度弹窗：创建一个包含QProgressBar的弹窗窗口，用于显示实时进度。
3. 信号槽连接：在主窗口的按钮点击事件中，启动子线程，将线程的进度信号连接到进度条的更新方法，线程结束信号连接到弹窗关闭和结果处理逻辑。
4. 修改耗时函数：让complicated_function能够在执行过程中发送进度信号（比如循环处理时按比例发送进度）。

修改后的完整代码 #

import sys
import pandas as pd
import time  # 仅用于模拟耗时操作，实际使用时删除

from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5 import QtCore, QtGui, QtWidgets


# 1. 定义子线程类，用于运行耗时函数
class WorkerThread(QThread):
    # 定义信号：传递进度百分比（int类型）
    progress_updated = pyqtSignal(int)
    # 定义信号：传递函数执行结果
    result_ready = pyqtSignal(object)

    def run(self):
        # 调用耗时函数，并传入进度回调
        result = self.complicated_function(self.update_progress)
        # 发送结果信号
        self.result_ready.emit(result)

    def complicated_function(self, progress_callback):
        # 模拟耗时操作：实际替换为你的业务逻辑
        total_steps = 100  # 假设总共有100个步骤
        for step in range(total_steps):
            # 这里执行实际的耗时任务（比如数据处理、文件读取等）
            time.sleep(0.05)  # 模拟耗时，实际删除
            # 计算进度百分比并发送
            progress = int((step + 1) / total_steps * 100)
            progress_callback(progress)
        
        # 返回实际结果，比如处理好的数据
        return "处理完成的结果"

    def update_progress(self, value):
        # 发送进度信号到主线程
        self.progress_updated.emit(value)


# 2. 定义进度弹窗类
class ProgressDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("处理进度")
        self.setFixedSize(300, 80)
        
        # 创建进度条
        self.progress_bar = QProgressBar(self)
        self.progress_bar.setGeometry(20, 20, 260, 30)
        self.progress_bar.setRange(0, 100)
        self.progress_bar.setValue(0)


class Ui_Dialog(object):
    def setupUi(self, Dialog):
        Dialog.setObjectName("Dialog")
        Dialog.resize(713, 511)
        self.tabWidget = QtWidgets.QTabWidget(Dialog)
        self.tabWidget.setGeometry(QtCore.QRect(20, 30, 671, 461))
        self.tabWidget.setObjectName("tabWidget")
        self.tab1 = QtWidgets.QWidget()
        self.tab1.setObjectName("tab1")
        font = QtGui.QFont()
        font.setPointSize(7)
        self.groupBox_Carga = QtWidgets.QGroupBox(self.tab1)
        self.groupBox_Carga.setEnabled(True)
        self.groupBox_Carga.setGeometry(QtCore.QRect(10, 90, 641, 161))
        self.groupBox_Carga.setObjectName("groupBox_Carga")
        self.toolButton = QtWidgets.QToolButton(self.groupBox_Carga)
        self.toolButton.setGeometry(QtCore.QRect(40, 30, 561, 31))
        self.toolButton.setObjectName("toolButton")
        self.groupBox_ConfMatcheo_4 = QtWidgets.QGroupBox(self.tab1)
        self.groupBox_ConfMatcheo_4.setEnabled(True)
        self.groupBox_ConfMatcheo_4.setGeometry(QtCore.QRect(10, 360, 641, 61))
        self.groupBox_ConfMatcheo_4.setFont(font)
        self.groupBox_ConfMatcheo_4.setObjectName("groupBox_ConfMatcheo_4")
        self.label = QtWidgets.QLabel(self.groupBox_ConfMatcheo_4)
        self.label.setGeometry(QtCore.QRect(80, 20, 521, 31))
        self.label.setObjectName("label")
        self.tabWidget.addTab(self.tab1, "")

        self.retranslateUi(Dialog)

    def retranslateUi(self, Dialog):
        _translate = QtCore.QCoreApplication.translate
        self.toolButton.setText(_translate("Dialog", "Load payments"))
        self.label.setText(_translate("Dialog", ""))

        self.toolButton.clicked.connect(self.press_button)

    def press_button(self):
        # 禁用按钮，防止重复点击
        self.toolButton.setEnabled(False)
        self.label.setText("正在处理，请稍候...")

        # 创建进度弹窗
        self.progress_dialog = ProgressDialog()
        self.progress_dialog.show()

        # 创建子线程
        self.worker_thread = WorkerThread()
        # 连接进度信号到进度条更新
        self.worker_thread.progress_updated.connect(self.progress_dialog.progress_bar.setValue)
        # 连接结果信号到结果处理函数
        self.worker_thread.result_ready.connect(self.handle_result)
        # 连接线程结束信号到清理操作
        self.worker_thread.finished.connect(self.cleanup_after_thread)

        # 启动线程
        self.worker_thread.start()

    def handle_result(self, result):
        # 处理耗时函数返回的结果
        self.label.setText(f"处理完成！结果：{result}")
        # 这里可以添加后续逻辑，比如将结果用于其他操作
        # variable = result

    def cleanup_after_thread(self):
        # 关闭进度弹窗
        self.progress_dialog.close()
        # 重新启用按钮
        self.toolButton.setEnabled(True)
        # 释放线程资源
        self.worker_thread.deleteLater()


if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    Dialog = QtWidgets.QMainWindow()
    ui = Ui_Dialog()
    ui.setupUi(Dialog)
    Dialog.show()
    sys.exit(app.exec_())

关键部分说明 #

• WorkerThread类：

  • 继承QThread，重写run()方法，在这里执行耗时函数。
  • progress_updated信号用于传递进度，result_ready信号传递最终结果。
  • complicated_function中通过回调函数progress_callback触发进度信号，你需要根据实际业务逻辑调整进度计算方式（比如按数据行数、任务阶段等）。

• ProgressDialog类：

  • 简单的弹窗，包含一个QProgressBar，用于显示实时进度。

• press_button方法：

  • 点击按钮后禁用按钮，防止重复触发。
  • 创建并显示进度弹窗，启动子线程，连接所有必要的信号槽。

• 信号槽机制：

  • 子线程不能直接操作UI，必须通过信号将数据发送到主线程，由主线程更新UI，这是PyQt的线程安全要求。

注意事项 #

1. 实际使用时，删除time.sleep(0.05)和模拟的循环逻辑，替换为你的真实complicated_function业务代码。
2. 如果你的耗时函数无法拆分步骤，可以考虑按时间估算进度，或者使用不确定进度条（progress_bar.setRange(0,0)）。
3. 确保线程结束后调用deleteLater()释放资源，避免内存泄漏。

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