Python Tkinter：如何使用线程来防止主事件循环“冻结”
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/16745507/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
Tkinter: How to use threads to preventing main event loop from "freezing"
提问by Dirty Penguin
I have a small GUI test with a "Start" button and a Progress bar. The desired behavior is:
我有一个带有“开始”按钮和进度条的小型 GUI 测试。所需的行为是：
- Click Start
- Progressbar oscillates for 5 seconds
- Progressbar stops
- 点击开始
- 进度条振荡 5 秒
- 进度条停止
The observed behavior is the "Start" button freezes for 5 seconds, then a Progressbar is displayed (no oscillation).
观察到的行为是“开始”按钮冻结 5 秒钟，然后显示进度条（无振荡）。
Here is my code so far:
到目前为止，这是我的代码：
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()
Based on the information from Bryan Oakley here, I understand that I need to use threads. I tried creating a thread, but I'm guessing that since the thread is started from within the main thread, it doesn't help.
根据此处来自 Bryan Oakley 的信息，我了解我需要使用线程。我尝试创建一个线程，但我猜因为该线程是从主线程中启动的，所以它没有帮助。
I had the idea to place the logic portion in a different class, and instantiate the GUI from within that class, similar to the example code by A. Rodas here.
我的想法是将逻辑部分放在不同的类中，并从该类中实例化 GUI，类似于 A. Rodas here的示例代码。
My question:
我的问题：
I can't figure out how to code it so that this command:
我无法弄清楚如何对其进行编码，以便此命令：
self.test_button = Button(self.master, command=self.tb_click)
calls a function that is located in the other class. Is this a Bad Thing to do or is it even possible? How would I create a 2nd class that can handle the self.tb_click? I tried following along to A. Rodas' example code which works beautifully. But I cannot figure out how to implement his solution in the case of a Button widget that triggers an action.
调用位于另一个类中的函数。这是一件坏事还是有可能？我将如何创建可以处理 self.tb_click 的第二个类？我尝试遵循 A. Rodas 的示例代码，它运行良好。但是我无法弄清楚如何在触发操作的 Button 小部件的情况下实现他的解决方案。
If I should instead handle the thread from within the single GUI class, how would one create a thread that doesn't interfere with the main thread?
如果我应该从单个 GUI 类中处理线程，那么如何创建一个不干扰主线程的线程？
采纳答案by A. Rodas
When you join the new thread in the main thread, it will wait until the thread finishes, so the GUI will block even though you are using multithreading.
当您在主线程中加入新线程时，它将等待线程完成，因此即使您使用多线程，GUI 也会阻塞。
If you want to place the logic portion in a different class, you can subclass Thread directly, and then start a new object of this class when you press the button. The constructor of this subclass of Thread can receive a Queue object and then you will be able to communicate it with the GUI part. So my suggestion is:
如果要将逻辑部分放在不同的类中，可以直接子类化Thread，然后按下按钮启动该类的新对象。Thread 的这个子类的构造函数可以接收一个 Queue 对象，然后您就可以将它与 GUI 部分进行通信。所以我的建议是：
- Create a Queue object in the main thread
- Create a new thread with access to that queue
- Check periodically the queue in the main thread
- 在主线程中创建一个Queue对象
- 创建一个可以访问该队列的新线程
- 定期检查主线程中的队列
Then you have to solve the problem of what happens if the user clicks two times the same button (it will spawn a new thread with each click), but you can fix it by disabling the start button and enabling it again after you call self.prog_bar.stop().
然后你必须解决如果用户点击同一个按钮两次会发生的问题（每次点击都会产生一个新线程），但是你可以通过禁用开始按钮并在调用后再次启用它来修复它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")
回答by jmihalicza
The problem is that t.join() blocks the click event, the main thread does not get back to the event loop to process repaints. See Why ttk Progressbar appears after process in Tkinteror TTK progress bar blocked when sending email
问题是 t.join() 阻塞了点击事件，主线程没有回到事件循环来处理重绘。请参阅为什么在 Tkinter 中的进程后出现 ttk进度条或发送电子邮件时 TTK 进度条被阻止
回答by BuvinJ
I will submit the basis for an alternate solution. It is not specific to a Tk progress bar per se, but it can certainly be implemented very easily for that.
我将提交替代解决方案的基础。它本身并不特定于 Tk 进度条，但当然可以很容易地实现它。
Here are some classes that allow you to run other tasks in the background of Tk, update the Tk controls when desired, and not lock up the gui!
这里有一些类允许您在 Tk 的后台运行其他任务，在需要时更新 Tk 控件，而不是锁定 gui！
Here's class TkRepeatingTask and BackgroundTask:
这是 TkRepeatingTask 和 BackgroundTask 类：
import threading
class 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()
Here's a Tk test which demos the use of these. Just append this to the bottom of the module with those classes in it if you want to see the demo in action:
这是一个 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()
Two import points I'll stress about BackgroundTask:
关于BackgroundTask，我要强调两个要点：
1) The function you run in the background task needs to take a function pointer it will both invoke and respect, which allows the task to be cancelled mid way through - if possible.
1）您在后台任务中运行的函数需要采用一个函数指针，它将调用和尊重，这允许在中途取消任务 - 如果可能的话。
2) You need to make sure the background task is stopped when you exit your application. That thread will still run even if your gui is closed if you don't address that!
2) 您需要确保退出应用程序时后台任务已停止。如果您不解决这个问题，即使您的 gui 已关闭，该线程仍将运行！
