CustomTkinter线程调用self.destroy()后特定场景下冻结问题排查

问题核心原因 #

1. 采集循环无法中断：_take_acquisition方法中的模拟采集循环一旦启动，会持续执行1秒且不检查停止标志，期间还会持续更新进度条；当窗口被销毁后，线程仍尝试操作已不存在的GUI控件，直接引发程序冻结。
2. GUI销毁时机过早：on_close方法中调用停止逻辑后立刻销毁GUI，没有等待线程完成当前采集任务并完全退出，导致线程与已销毁的GUI资源产生冲突。

解决方案 #

关键改动点 #

1. 让采集循环可被中断：在嵌套循环中加入停止标志检查，允许中途退出采集。
2. 保护GUI控件操作：线程更新进度条前，先判断窗口是否处于关闭状态，避免操作已销毁的控件。
3. 等待线程完全退出后再销毁GUI：使用CustomTkinter的after方法轮询线程状态，避免在GUI线程中直接调用join导致阻塞。

修正后的完整代码 #

import customtkinter as ctk
from threading import Thread, Event
import time
from enum import IntEnum


class ThreadState(IntEnum):
    READY = 0
    CONTINUOUS = 1


class MainApp(ctk.CTk):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.stop_event = Event()
        self.data_thread = None
        self.is_closing = False  # 标记窗口是否正在关闭

        self.title("Dummy Program")
        self.geometry("310x75")

        self.control_frame = ctk.CTkFrame(self, fg_color="transparent")
        self.control_frame.pack(side="top", fill="both", padx=10, pady=10)

        start_button = ctk.CTkButton(self.control_frame, text="Start Acquisition", command=self.start_continuous)
        start_button.pack(side="left")

        stop_button = ctk.CTkButton(self.control_frame, text="Stop Acquisition", command=self.stop_continuous)
        stop_button.pack(side="right")

        self.progressbar = ctk.CTkProgressBar(self)
        self.progressbar.pack(side="bottom", fill="both", padx=10, pady=10)

        self.protocol("WM_DELETE_WINDOW", self.on_close)

    def start_continuous(self):
        if not self.data_thread or not self.data_thread.is_alive():
            self.stop_event.clear()
            self.data_thread = DataThread(self, self.stop_event)
            self.data_thread.start()
            self.data_thread.acquire_continuous_start()

    def stop_continuous(self):
        if self.data_thread and self.data_thread.is_alive():
            self.stop_event.set()
            self.data_thread.acquire_continuous_stop()

    def on_close(self):
        print('Stopping thread @ ', time.time())
        self.stop_continuous()
        self.is_closing = True
        # 轮询线程状态，直到线程退出再销毁GUI
        self.check_thread_exit()

    def check_thread_exit(self):
        if self.data_thread and self.data_thread.is_alive():
            self.after(100, self.check_thread_exit)  # 每100ms检查一次
        else:
            print('Closing GUI @ ', time.time())
            self.destroy()
            print('程序已正常退出')


class DataThread(Thread):
    def __init__(self, parent_app, stop_event):
        super(DataThread, self).__init__()
        self.parent_app: MainApp = parent_app
        self.stop_event = stop_event
        self.get_data = False
        self.thread_state = ThreadState.READY

    def run(self):
        while not self.stop_event.is_set():
            match self.thread_state:
                case ThreadState.CONTINUOUS:
                    self._continuous_state()
                    self.thread_state = ThreadState.READY
                case ThreadState.READY:
                    print("Ready")
                    time.sleep(0.5)
                case _:
                    print("Unknown state")
                    time.sleep(0.5)
        print("采集线程已退出")

    def _take_acquisition(self):
        print("Acquiring dummy data")
        start_time = time.time()
        elapsed_time = 0.0
        while elapsed_time < 1.0:
            # 检查停止标志，允许中途退出采集
            if self.stop_event.is_set() or not self.get_data:
                break
            time.sleep(0.05)
            elapsed_time = time.time() - start_time
            # 窗口未关闭时才更新进度条
            if not self.parent_app.is_closing:
                self.parent_app.progressbar.set(max(min(1.0, elapsed_time), 0.0))

    def _continuous_state(self):
        # 双重检查停止标志，确保线程能及时响应退出信号
        while self.get_data and not self.stop_event.is_set():
            self._take_acquisition()

    def acquire_continuous_start(self):
        self.get_data = True
        self.thread_state = ThreadState.CONTINUOUS

    def acquire_continuous_stop(self):
        self.get_data = False


def main():
    app = MainApp()
    app.mainloop()


if __name__ == "__main__":
    main()

效果验证 #

• 触发冻结流程：点击“Start Acquisition”后直接关闭窗口，程序会先等待采集线程完成当前任务或响应停止信号，之后优雅退出，不再冻结。
• 正常退出流程：点击停止按钮后关闭窗口，依然保持正常退出逻辑。

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