PyQt6中如何等待所有创建的QThread执行完毕后再运行后续代码

直接利用QThread自带的finished信号做完成计数即可，不需要自己写轮询查状态的逻辑，没有额外性能开销，也不会卡线程。

先提你现有代码里两个容易出问题的点：

• 别用QThread.terminate()强杀线程，这个方法会直接中断线程执行，不会做任何资源清理，很容易导致程序崩溃、内存泄漏，后面的实现会改成优雅退出的逻辑
• 不要写死循环轮询每个线程的isFinished()状态，会平白占用CPU资源，Qt的信号槽机制本身就是为这种场景设计的，触发延迟在微秒级

实现思路 #

• 在DataCollectionLoop类里新增三个成员：完成线程计数器、单批次总线程数、优雅退出标志位
• 每启动一批子线程前，重置计数器为0，记录本次要启动的线程总数
• 给每个启动的子线程绑定finished信号，每有一个线程执行完run()方法，就自动触发计数+1
• 当计数器数值等于当前批次总线程数时，自动执行你需要等所有线程跑完才运行的后续逻辑，清理已完成的线程对象，等待设定的10秒间隔后再启动下一批任务
• 窗口关闭时把退出标志位置为True，等当前批次任务执行完自动退出工作线程，不做强杀

修改后的可运行代码 #

from PyQt6.QtCore import QThread, QObject
from PyQt6.QtWidgets import QApplication, QMainWindow, QWidget
import time


class GetMarketDataV2(QThread):

    def __init__(self, index):
        super(GetMarketDataV2, self).__init__()
        self.index = index

    def run(self):
        time.sleep(5)
        print(f"Thread: {self.index}\n")


class DataCollectionLoop(QObject):
    """
    Runs as long as the GUI is running. Terminated with the Main window close event.
    The task of this class is, to start a specific amount of threads with a specific interval.
    """
    def __init__(self):
        super(DataCollectionLoop, self).__init__()
        self.threads = {}
        self.finished_count = 0
        self.total_threads_per_batch = 10
        self.is_running = True

    def start_one_batch(self):
        # 重置计数
        self.finished_count = 0
        # 批量启动子线程
        for i in range(self.total_threads_per_batch):
            thread = GetMarketDataV2(index=i)
            thread.finished.connect(self.on_single_thread_finished)
            self.threads[i] = thread
            thread.start()

    def on_single_thread_finished(self):
        self.finished_count += 1
        # 判断所有线程是否都执行完成
        if self.finished_count == self.total_threads_per_batch:
            # 这里放所有线程跑完后要执行的逻辑
            print("print this statement after all threads are finished")
            
            # 清理已完成的线程对象，避免内存泄漏
            for t in self.threads.values():
                t.deleteLater()
            self.threads.clear()

            # 处于运行状态的话，等10秒启动下一批
            if self.is_running:
                QThread.msleep(10000)
                self.start_one_batch()

    def run(self):
        # 启动第一批任务
        self.start_one_batch()
        # 启动线程本地事件循环，保证信号槽能正常触发
        QThread.currentThread().exec()

    def stop(self):
        self.is_running = False
        QThread.currentThread().quit()


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()

        # 初始化数据采集工作线程
        self.data_thread = QThread()

        # 实例化循环采集对象并移动到工作线程
        self.data_collection = DataCollectionLoop()
        self.data_collection.moveToThread(self.data_thread)

        # 线程启动时触发采集逻辑
        self.data_thread.started.connect(self.data_collection.run)

        # 启动工作线程
        self.data_thread.start()

        # 最小化GUI配置
        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)
        self.show()

    # 优雅关闭线程，不用terminate强杀
    def closeEvent(self, event):
        self.data_collection.stop()
        self.data_thread.wait()
        event.accept()


app = QApplication([])
window = MainWindow()
app.exec()

补充说明 #

• 这个实现没有任何阻塞轮询的逻辑，线程完成后信号自动触发，计数到点立刻执行后续逻辑，性能损耗可以忽略
• 所有线程都是正常执行完退出，没有强制中断的操作，不会出现资源泄漏或者程序崩溃的问题
• 如果后续需要在所有子线程完成后更新GUI内容，不要直接在DataCollectionLoop的工作线程里操作控件，自定义一个信号发给主线程执行GUI更新即可，否则会触发Qt的跨线程访问报错

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