Python-Tkinter：如何使用线程防止主事件循环“冻结”

Question

我有一个带有“开始”按钮和进度条的小型GUI测试。所需的行为是：

• 点击开始
• 进度条振荡5秒钟
• 进度栏停止

观察到的行为是“开始”按钮冻结5秒钟，然后显示进度条（无振荡）。

到目前为止，这是我的代码：

class GUI:
    def __init__(self, master):
        self.master = master
        self.test_button = Button(self.master, command=self.tb_click)
        self.test_button.configure(
            text="Start", background="Grey",
            padx=50
            )
        self.test_button.pack(side=TOP)

    def progress(self):
        self.prog_bar = ttk.Progressbar(
            self.master, orient="horizontal",
            length=200, mode="indeterminate"
            )
        self.prog_bar.pack(side=TOP)

    def tb_click(self):
        self.progress()
        self.prog_bar.start()
        # Simulate long running process
        t = threading.Thread(target=time.sleep, args=(5,))
        t.start()
        t.join()
        self.prog_bar.stop()

root = Tk()
root.title("Test Button")
main_ui = GUI(root)
root.mainloop()

根据Bryan Oakley 在此提供的信息，我了解我需要使用线程。我尝试创建一个线程，但是我猜测由于该线程是从主线程中启动的，因此没有帮助。

我有想法放置在不同的类中的逻辑部分，以及从该类，类似于由A.罗达斯示例代码内实例化GUI 这里。

我的问题：

我不知道如何编码，以便此命令：

self.test_button = Button(self.master, command=self.tb_click)

调用另一个类中的函数。这是一件坏事吗，甚至有可能吗？我将如何创建可以处理self.tb_click的第二个类？我尝试遵循A. Rodas的示例代码，该示例代码运行良好。但是我无法弄清楚在按钮小部件触发动作的情况下如何实现他的解决方案。

如果我应该从单个GUI类中处理线程，那么如何创建一个不会干扰主线程的线程？

Answer 1

当您将新线程加入主线程时，它将等待直到线程完成，因此即使您正在使用多线程，GUI也会阻塞。

如果要将逻辑部分放在其他类中，则可以直接将Thread子类化，然后在按下按钮时启动该类的新对象。Thread的此子类的构造函数可以接收Queue对象，然后您就可以与GUI部件进行通信。所以我的建议是：

• 在主线程中创建一个队列对象
• 创建一个可以访问该队列的新线程
• 定期检查主线程中的队列

然后，您必须解决以下问题：如果用户两次单击同一按钮两次（每次单击都会产生一个新线程），但是可以通过禁用开始按钮并在调用后再次启用它来解决此问题self.prog_bar.stop()。

import Queue

class GUI:
    # ...

    def tb_click(self):
        self.progress()
        self.prog_bar.start()
        self.queue = Queue.Queue()
        ThreadedTask(self.queue).start()
        self.master.after(100, self.process_queue)

    def process_queue(self):
        try:
            msg = self.queue.get(0)
            # Show result of the task if needed
            self.prog_bar.stop()
        except Queue.Empty:
            self.master.after(100, self.process_queue)

class ThreadedTask(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
    def run(self):
        time.sleep(5)  # Simulate long running process
        self.queue.put("Task finished")