如何实现含多线程、多进程的Tkinter应用优雅终止？

如何让Python tkinter应用在正常运行结束时，以及用户终止窗口中断正在执行的处理时都能优雅关闭？

该tkinter应用点击按钮后启动长时间运行的函数long.work()，此函数在新线程中启动，以保证tkinter应用在长时执行期间保持响应。

work()中调用的部分函数通过ThreadPoolExecutor实现多线程（concurrent_tasks.do_multi_threads()），其他函数则通过joblib的Parallel()和delayed()实现多进程（parallel_tasks.do_multi_process()）。

因此，优雅终止需处理以下场景：无用户终止、用户在函数完成后终止、用户在多线程/多进程阶段中断，同时需关闭使用QueueHandler和QueueListener写入日志文件的日志器。

我已查看Stack Overflow上的类似问题，但未找到覆盖所有所需场景的方案。

目前我的实现方案在用户关闭窗口中断多线程函数时会报错（其他场景也存在问题）：

app.py #

import logging
from logging.handlers import QueueHandler, QueueListener
import sys
import multiprocessing
import threading
import tkinter as tk
from tkinter import Tk, ttk
from joblib.externals.loky import get_reusable_executor

import long


def finish(
    logger: logging.Logger,
    listener: QueueListener = None,
    log_queue: multiprocessing.Queue = None,
    threads: bool = False,
    lock: threading.Lock = None,
    root: tk.Tk = None,
    exit_code: int = 0,
):
    # QueueHandler
    if log_queue is not None:
        log_queue.put_nowait(None)

    # QueueListener
    if listener is not None:
        listener.stop()

    # joblib
    get_reusable_executor(kill_workers=True).shutdown(wait=True)

    # multiprocessing
    for p in multiprocessing.active_children():
        p.terminate()
        p.join()

    # threading
    if threads is True:
        for thread in threading.enumerate():
            if thread is not threading.current_thread():
                thread.join(timeout=1)

    # logger
    for handler in logger.handlers:
        handler.flush()
        handler.close()
    logger.handlers.clear()

    # lock
    if lock is not None and lock.locked():
        lock.release()

    # tkinter
    if root is not None:
        root.destroy()
        sys.exit(exit_code)


def main():
    root = Tk()

    # logger
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    log_fmt = logging.Formatter(
        "%(asctime)s (%(levelname)s) %(message)s", datefmt="%H:%M:%S"
    )
    file_handler = logging.FileHandler("log_file.log")
    file_handler.setFormatter(log_fmt)
    file_handler.setLevel(logging.DEBUG)
    
    # 注意：QueueListener还添加了一个StreamHandler

    manager = multiprocessing.Manager()
    log_queue = manager.Queue()
    queue_handler = QueueHandler(log_queue)
    logger.addHandler(queue_handler)

    listener = QueueListener(log_queue, file_handler, respect_handler_level=True)
    listener.start()

    def begin():
        # entry_vals = 获取输入框的值...
        exit_code = 0
        lock = threading.Lock()

        def terminate():
            finish(logger, listener, log_queue, threads=True, lock=lock, root=root, exit_code=exit_code)
        
        root.protocol("WM_DELETE_WINDOW", terminate)

        def start():  # 编辑补充
            try:
                long.work(entry_vals)
            finally:
                button.config(state=tk.NORMAL)  # 重新启用按钮
                finish(logger, listener, log_queue, threads=True, lock=lock, root=None, exit_code=exit_code)

        try:  # 编辑补充（缩进调整）
            set_progress = entry_vals.get("update_status")
            if set_progress:  # 编辑补充
                set_progress("begin", 1)  # 编辑补充
            logging.debug("启动GUI...")
            button.config(state=tk.DISABLED)  # 禁用按钮
            threading.Thread(target=start).start()  # 编辑补充
        except Exception as e:
            exit_code = 1
            logging.exception(e)
            finish(logger, listener, log_queue, threads=False, lock=None, root=None, exit_code=exit_code)

    frame = ttk.Frame(root)
    frame.grid(row=0, column=0, sticky="ew", padx=5, pady=5)
    button = ttk.Button(frame, text="开始", command=begin)

    root.mainloop()

if __name__ == "__main__":
    main()

long.py #

import logging
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from joblib import Parallel, delayed

import concurrent_tasks
import parallel_tasks

def work(entry_vals):
    set_progress = entry_vals.get("update_status")
    partial_func = partial(concurrent_tasks.do_multi_threads, entry_vals)
    if set_progress:  # 编辑补充
        set_progress("work", 10)  # 编辑补充
    logging.debug("启动多线程函数...")  # 编辑补充
    with ThreadPoolExecutor() as executor:
        res_1 = list(executor.map(partial_func, range(10)))
    
    if set_progress:  # 编辑补充
        set_progress("work", 60)  # 编辑补充
    logging.debug("启动多进程函数...")  # 编辑补充
    with Parallel(n_jobs=-1) as parallel_work:
        res_2 = list(parallel_work(delayed(parallel_tasks.do_multi_process)(entry_vals, i) for i in range(10)))

    return res_1, res_2

本次编辑补充了简化代码的细节：通过函数式编程，设置tk.IntVar来递增tk.ttk.Progressbar，该变量是Progressbar参数variable的实参。

app.py #

def update_status(
    progress_var,
    progress_bar,
    percent_label,
    status_label,
    lock
):
    def callback(section, percent):
        with lock:
            if percent is not None:
                progress_var.set(percent)
            percent_label["text"] = f"{progress_var.get()}%"
            status_label["text"] = section

            progress_bar.update_idletasks()
            percent_label.update_idletasks()
            status_label.update_idletasks()

        progress_bar.after(10000, callback)
    return callback

错误信息 #

Exception has occurred: RuntimeError
main thread is not in main loop

in start:
long.work(entry_vals)

in work:
set_progress("work", 60)

in callback:
progress_var.set(percent)
RuntimeError: main thread is not in main loop

During handling of the above exception, another exception occurred:
in start:
button.config(state=tk.NORMAL)
RuntimeError: main thread is not in main loop

核心问题分析 #

报错的根本原因是非主线程直接操作Tkinter组件。Tkinter的GUI操作必须在启动mainloop的主线程中执行，子线程调用set_progress修改进度条、button.config修改按钮状态都会触发RuntimeError。此外，当前的终止逻辑属于强制中断，未让任务主动退出，容易导致资源泄漏。

针对性修复步骤 #

1. 用root.after将GUI操作切换到主线程
   所有修改Tkinter组件的操作，都要通过root.after提交到主线程执行：

   # 修改update_status的callback
   def update_status(
       progress_var,
       progress_bar,
       percent_label,
       status_label,
       lock,
       root  # 新增：传入主窗口对象
   ):
       def callback(section, percent):
           def update_gui():
               with lock:
                   if percent is not None:
                       progress_var.set(percent)
                   percent_label["text"] = f"{progress_var.get()}%"
                   status_label["text"] = section
   
                   progress_bar.update_idletasks()
                   percent_label.update_idletasks()
                   status_label.update_idletasks()
           # 提交到主线程执行
           root.after(0, update_gui)
       return callback
   

   在start函数中修改按钮状态：

   def start():
       try:
           long.work(entry_vals)
       finally:
           # 用after切换到主线程启用按钮
           root.after(0, lambda: button.config(state=tk.NORMAL))
           finish(logger, listener, log_queue, threads=True, lock=lock, root=None, exit_code=exit_code)
   

2. 添加共享终止标志，实现任务主动退出
   新增跨进程/线程的布尔标志，让多线程/多进程任务定期检测并主动终止：

   # 在main函数中添加共享终止标志
   from multiprocessing import Value
   
   def main():
       root = Tk()
       # 跨进程/线程的终止标志
       should_terminate = Value('b', False)
       # ...其他代码
   
       def terminate():
           # 设置终止标志
           should_terminate.value = True
           finish(logger, listener, log_queue, threads=True, lock=lock, root=root, exit_code=exit_code)
   

   在long.work中检测标志：

   def work(entry_vals, should_terminate):  # 新增参数
       set_progress = entry_vals.get("update_status")
       partial_func = partial(concurrent_tasks.do_multi_threads, entry_vals, should_terminate)
       if set_progress:
           set_progress("work", 10)
       logging.debug("启动多线程函数...")
       with ThreadPoolExecutor() as executor:
           res_1 = []
           for result in executor.map(partial_func, range(10)):
               if should_terminate.value:
                   executor.shutdown(wait=False)
                   raise RuntimeError("用户终止任务")
               res_1.append(result)
       
       if should_terminate.value:
           raise RuntimeError("用户终止任务")
       
       if set_progress:
           set_progress("work", 60)
       logging.debug("启动多进程函数...")
       entry_vals["should_terminate"] = should_terminate
       with Parallel(n_jobs=-1) as parallel_work:
           res_2 = []
           for result in parallel_work(delayed(parallel_tasks.do_multi_process)(entry_vals, i) for i in range(10)):
               if should_terminate.value:
                   parallel_work._backend.shutdown()
                   raise RuntimeError("用户终止任务")
               res_2.append(result)
   
       return res_1, res_2
   

3. 完善finish函数的资源清理逻辑
   优化资源清理步骤，避免阻塞主线程和重复操作：

   def finish(
       logger: logging.Logger,
       listener: QueueListener = None,
       log_queue: multiprocessing.Queue = None,
       worker_threads: list = None,  # 改为传入具体工作线程列表
       lock: threading.Lock = None,
       root: tk.Tk = None,
       exit_code: int = 0,
   ):
       # QueueHandler
       if log_queue is not None:
           log_queue.put_nowait(None)
   
       # QueueListener
       if listener is not None:
           listener.stop()
           listener.join()
   
       # joblib
       get_reusable_executor(kill_workers=True).shutdown(wait=True)
   
       # multiprocessing
       for p in multiprocessing.active_children():
           p.terminate()
           p.join()
   
       # threading：只join自己启动的工作线程
       if worker_threads is not None:
           for thread in worker_threads:
               if thread.is_alive():
                   thread.join(timeout=2)
   
       # logger：先移除处理器再关闭
       while logger.handlers:
           handler = logger.handlers.pop()
           handler.flush()
           handler.close()
   
       # lock
       if lock is not None and lock.locked():
           lock.release()
   
       # tkinter：仅在主线程销毁窗口
       if root is not None and threading.current_thread() is threading.main_thread():
           root.destroy()
           sys.exit(exit_code)
   

完整流程梳理 #

1. 用户点击按钮后，主线程启动工作线程，同时设置窗口关闭的回调函数
2. 工作线程执行long.work，多线程/多进程任务定期检测终止标志
3. 任务正常完成时，工作线程通过root.after通知主线程更新按钮状态，随后调用finish清理资源
4. 用户关闭窗口时，回调函数设置终止标志，调用finish终止所有子进程/线程，完成日志和GUI资源清理

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