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

Question

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

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

• 单击“开始”
• Proressbar振荡5秒
• Progressbar停止

观察到的行为是“开始”按钮冻结5秒，然后显示一个Progressbar(没有振荡)。

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

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()

根据布莱恩·奥克利的信息这里我知道我需要使用线程。我尝试创建一个线程，但我猜想，由于线程是从主线程内部启动的，所以没有帮助。

我的想法是将逻辑部分放在不同的类中，并从该类中实例化GUI，类似于A.Rodas的示例代码这里.

我的问题是：

我不知道如何对它进行编码，以便这个命令：

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

调用位于另一个类中的函数。这是一件坏事，还是甚至有可能？我将如何创建第二个类来处理Sel.tb_Click？我试着遵循A.Rodas的示例代码，它工作得很好。但我不知道如何在触发操作的Button小部件的情况下实现他的解决方案。

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

Answer 1

当您在主线程中加入新线程时，它将等待线程完成，所以即使您正在使用多线程，GUI也会阻塞。如果要将逻辑部分放置在不同的类中，可以直接子类线程，然后在按下按钮时启动该类的新对象。Thread子类的构造函数可以接收队列对象，然后您将能够与GUI部分通信。所以我的建议是：在主线程中创建队列对象创建一个访问该队列的新线程。定期检查主线程中的队列然后，您必须解决这样的问题：如果用户单击同一按钮两次(每次单击将生成一个新线程)，则会发生什么情况，但是您可以通过禁用开始按钮并在调用后再次启用它来修复它。self.prog_bar.stop().import Queueclass 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")

Answer 2

问题是t.join()阻塞了Click事件，主线程没有返回到Event循环来处理重新绘制。看见TTK进程在Tkinter过程后出现的原因或发送电子邮件时阻止TTK进度条

Answer 3

我将提出另一种解决办法的基础。它并不是特定于Tk进度条本身，但它当然可以很容易地实现。下面的一些类允许您在Tk的背景下运行其他任务，在需要时更新Tk控件，而不锁定GUI！下面是TkRepeatingTask和BackoundTask的类：import threadingclass TkRepeatingTask():     def __init__( self, tkRoot, taskFuncPointer, freqencyMillis ):         self.__tk_   = tkRoot         self.__func_ = taskFuncPointer                 self.__freq_ = freqencyMillis         self.__isRunning_ = False     def isRunning( self ) : return self.__isRunning_      def start( self ) :          self.__isRunning_ = True         self.__onTimer()     def stop( self ) : self.__isRunning_ = False     def __onTimer( self ):          if self.__isRunning_ :             self.__func_()              self.__tk_.after( self.__freq_, self.__onTimer )class BackgroundTask():     def __init__( self, taskFuncPointer ):         self.__taskFuncPointer_ = taskFuncPointer         self.__workerThread_ = None         self.__isRunning_ = False     def taskFuncPointer( self ) : return self.__taskFuncPointer_    def isRunning( self ) :          return self.__isRunning_ and self.__workerThread_.isAlive()     def start( self ):          if not self.__isRunning_ :             self.__isRunning_ = True             self.__workerThread_ = self.WorkerThread( self )             self.__workerThread_.start()     def stop( self ) : self.__isRunning_ = False     class WorkerThread( threading.Thread ):         def __init__( self, bgTask ):                   threading.Thread.__init__( self )             self.__bgTask_ = bgTask        def run( self ):             try :                 self.__bgTask_.taskFuncPointer()( self.__bgTask_.isRunning )             except Exception as e: print repr(e)             self.__bgTask_.stop()这里是一个Tk测试，演示了这些测试的使用。如果您想要查看实际操作中的演示，只需将其附加到模块底部，其中包含这些类：def tkThreadingTest():     from tkinter import Tk, Label, Button, StringVar     from time import sleep    class UnitTestGUI:         def __init__( self, master ):             self.master = master             master.title( "Threading Test" )             self.testButton = Button(                  self.master, text="Blocking", command=self.myLongProcess )             self.testButton.pack()             self.threadedButton = Button(                  self.master, text="Threaded", command=self.onThreadedClicked )             self.threadedButton.pack()             self.cancelButton = Button(                  self.master, text="Stop", command=self.onStopClicked )             self.cancelButton.pack()             self.statusLabelVar = StringVar()             self.statusLabel = Label( master, textvariable=self.statusLabelVar )             self.statusLabel.pack()             self.clickMeButton = Button(                  self.master, text="Click Me", command=self.onClickMeClicked )             self.clickMeButton.pack()             self.clickCountLabelVar = StringVar()                         self.clickCountLabel = Label( master,  textvariable=self.clickCountLabelVar )             self.clickCountLabel.pack()             self.threadedButton = Button(                  self.master, text="Timer", command=self.onTimerClicked )             self.threadedButton.pack()             self.timerCountLabelVar = StringVar()                         self.timerCountLabel = Label( master,  textvariable=self.timerCountLabelVar )             self.timerCountLabel.pack()             self.timerCounter_=0             self.clickCounter_=0             self.bgTask = BackgroundTask( self.myLongProcess )             self.timer = TkRepeatingTask( self.master, self.onTimer, 1 )         def close( self ) :             print "close"             try: self.bgTask.stop()             except: pass             try: self.timer.stop()             except: pass                         self.master.quit()         def onThreadedClicked( self ):             print "onThreadedClicked"             try: self.bgTask.start()             except: pass         def onTimerClicked( self ) :             print "onTimerClicked"             self.timer.start()         def onStopClicked( self ) :             print "onStopClicked"             try: self.bgTask.stop()             except: pass             try: self.timer.stop()             except: pass                                 def onClickMeClicked( self ):             print "onClickMeClicked"             self.clickCounter_+=1             self.clickCountLabelVar.set( str(self.clickCounter_) )         def onTimer( self ) :             print "onTimer"             self.timerCounter_+=1             self.timerCountLabelVar.set( str(self.timerCounter_) )         def myLongProcess( self, isRunningFunc=None ) :             print "starting myLongProcess"             for i in range( 1, 10 ):                 try:                     if not isRunningFunc() :                         self.onMyLongProcessUpdate( "Stopped!" )                         return                 except : pass                    self.onMyLongProcessUpdate( i )                 sleep( 1.5 ) # simulate doing work             self.onMyLongProcessUpdate( "Done!" )                         def onMyLongProcessUpdate( self, status ) :             print "Process Update: %s" % (status,)             self.statusLabelVar.set( str(status) )     root = Tk()         gui = UnitTestGUI( root )     root.protocol( "WM_DELETE_WINDOW", gui.close )     root.mainloop()if __name__ == "__main__":      tkThreadingTest()关于背景任务，我将强调两个要点：1)在后台任务中运行的函数需要接受它将调用和尊重的函数指针，这允许在执行过程中取消任务-如果可能的话。2)在退出应用程序时，需要确保后台任务被停止。即使您的gui被关闭，如果您不解决这个问题，该线程仍然会运行！