PyQt6中QThreadPool与QThread的选择与正确关闭策略

from PyQt6.QtCore import QThread, pyqtSignal
from typing import Callable, Any

class TaskRunner(QThread):
    # 定义一个信号，用于在任务完成后通知主线程
    finished_signal = pyqtSignal()

    def __init__(self, parent: Any | None, task: Callable):
        super().__init__()
        self.parent = parent
        self.task = task

    def run(self):
        """
        在新的线程中执行耗时任务。
        """
        try:
            self.task(self.parent)
        finally:
            # 任务完成后发出信号
            self.finished_signal.emit()

from src.gui.loading import Ui_Form # 假设Ui_Form是你的UI定义
from PyQt6.QtWidgets import QWidget, QApplication
from PyQt6.QtCore import QTimer, QThread, pyqtSignal
from typing import Callable, Any

class Loading(Ui_Form, QWidget):
    def __init__(self,
                 parent: QWidget | None,
                 next_widget: QWidget | None,
                 action: str,
                 time: int,
                 task: Callable,
                 task_len: int,
                 initial_task: str):
        super().__init__()
        self.setupUi(self)
        self.setParent(parent)
        self.parent = parent
        self.next_widget = next_widget
        self.time = time

        # 直接实例化 TaskRunner (QThread)
        self.task_thread = TaskRunner(self, task)
        # 连接任务完成信号
        self.task_thread.finished_signal.connect(self.on_task_finished)

        self.current_time = 0
        self.tasks_done = 0
        self.all_tasks = task_len

        self.Task.setText(action)
        self.Estimation.setText(f"estimated time: {self.int_to_time(time)}")
        self.progressBar.setValue(0)
        self.TimeLeft.setText("")
        self.Current.setText("")
        self.Task.setText("")

        # 启动任务线程
        self.run_tasks()
        self.task_done(initial_task)

        self.timer = QTimer(self)
        self.timer.timeout.connect(self.update_time)
        self.timer.start(1000)

    @staticmethod
    def int_to_time(time: int) -> str:
        if time >= 3600:
            return f"{time / 3600} hours"
        elif time >= 60:
            return f"{time / 60} minutes"
        else:
            return f"{time} seconds"

    def update_time(self):
        self.current_time += 1
        self.TimeLeft.setText(self.int_to_time(self.current_time))

    def task_done(self, next_task: str = None):
        # 这里的tasks_done逻辑可能需要根据实际任务数量调整
        # 如果只有一个大任务，那么只在on_task_finished中更新一次即可
        self.tasks_done += 1
        if not next_task:
            self.Current.setText("finished all tasks, closing window")
            self.Tasks.setText(f"{self.tasks_done} out of {self.all_tasks}")
            # 任务完成后，可以更新进度条到100%
            self.progressBar.setValue(100) 
        elif self.tasks_done != self.all_tasks:
            self.Current.setText(f"currently: {next_task}")
            self.Tasks.setText(f"{self.tasks_done} out of {self.all_tasks}")
            # 如果有多个子任务，这里可以根据tasks_done更新进度条
            self.progressBar.setValue(int(self.tasks_done / self.all_tasks * 100))

    def on_task_finished(self):
        """
        当后台任务完成后，此槽函数会被调用。
        """
        self.task_done(None) # 标记所有任务完成
        # 停止计时器
        self.timer.stop()
        # 确保线程正确退出
        self.task_thread.quit()
        self.task_thread.wait() # 阻塞直到线程完全退出
        self.close() # 关闭窗口

    def closeEvent(self, event):
        """
        重写closeEvent以确保在窗口关闭时线程和计时器都被正确停止。
        """
        self.timer.stop()
        if self.task_thread.isRunning():
            self.task_thread.quit()
            self.task_thread.wait()
        super().closeEvent(event) # 调用父类的closeEvent

    def run_tasks(self):
        # 直接启动 QThread
        self.task_thread.start()

import time
from unittest import TestCase
from PyQt6.QtWidgets import QApplication

# 假设 Loading 类和 Ui_Form 都在可导入的路径中
# from your_module import Loading, Ui_Form 

class TestLoading(TestCase):
    def test_task(self):
        def foo(loading_page: Loading):
            # 模拟耗时操作
            time.sleep(5)
            # 可以在这里更新进度或状态，通过信号发送回主线程
            # loading_page.update_progress_signal.emit(50) 
            time.sleep(5) # 模拟更多工作

        app = QApplication([])
        # 注意：这里的 task_len 应该与 foo 函数中模拟的实际任务阶段数匹配，
        # 如果 foo 只是一个整体任务，那么 task_len 可以设为1，或者根据你的需求调整
        self.loader = Loading(None, None, "doing something", 10, foo, 1, "testing") 
        self.loader.show()
        app.exec()