Tkinter：如何使用線程防止主事件循環(huán)“凍結(jié)”我有一個帶有“開始”按鈕和進(jìn)度條的小型GUI測試。期望的行為是：單擊“開始”Proressbar振蕩5秒Progressbar停止觀察到的行為是“開始”按鈕凍結(jié)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()根據(jù)布萊恩·奧克利的信息這里我知道我需要使用線程。我嘗試創(chuàng)建一個線程，但我猜想，由于線程是從主線程內(nèi)部啟動的，所以沒有幫助。我的想法是將邏輯部分放在不同的類中，并從該類中實例化GUI，類似于A.Rodas的示例代碼這里.我的問題是：我不知道如何對它進(jìn)行編碼，以便這個命令：self.test_button = Button(self.master, command=self.tb_click)調(diào)用位于另一個類中的函數(shù)。這是一件壞事，還是甚至有可能？我將如何創(chuàng)建第二個類來處理Sel.tb_Click？我試著遵循A.Rodas的示例代碼，它工作得很好。但我不知道如何在觸發(fā)操作的Button小部件的情況下實現(xiàn)他的解決方案。如果我應(yīng)該從單個GUI類中處理線程，那么如何創(chuàng)建一個不干擾主線程的線程呢？
3 回答
慕的地10843
TA貢獻(xiàn)1785條經(jīng)驗 獲得超8個贊
在主線程中創(chuàng)建隊列對象 創(chuàng)建一個訪問該隊列的新線程。 定期檢查主線程中的隊列
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")
慕哥9229398
TA貢獻(xiàn)1877條經(jīng)驗 獲得超6個贊
HUH函數(shù)
TA貢獻(xiàn)1836條經(jīng)驗 獲得超4個贊
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()
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()
添加回答
舉報
0/150
提交
取消