python3 tkinter threading ( tkinter에서 thread사용 )

환경
언어 : Python 3.7.3
GUI : Tkinter

뭔가 복잡한고 오래걸리는 작업을 하면 GUI가 멈춥니다. thread로 만들어야 하며 예제입니다.

기본 내용은 아래 링크입니다.

https://stackoverflow.com/questions/16745507/tkinter-how-to-use-threads-to-preventing-main-event-loop-from-freezing

python 2 내용이라 3에 맞춰서 코드 수정하였고, 불필요하다고 생각되는 부분을 삭제하여 좀 더 쉽게 예제를 만들었습니다.

import threading

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.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.bgTask = BackgroundTask( self.myLongProcess )

        def close( self ) :
            print ("close")
            try: self.bgTask.stop()
            except: pass
            self.master.quit()

        def onThreadedClicked( self ):
            print ("onThreadedClicked")
            try: self.bgTask.start()
            except: pass

        def onStopClicked( self ) :
            print ("onStopClicked")
            try: self.bgTask.stop()
            except: pass

        def myLongProcess( self, isRunningFunc=None ) :
            print ("starting myLongProcess")
            # WORKING THREAD
            for i in range( 1, 10 ):
                try:
                    if not isRunningFunc() :
                        self.onMyLongProcessUpdate( "Stopped!" )
                        return
                except : pass   
                self.onMyLongProcessUpdate( i )
                # HERE
                # HERE
                # HERE
                sleep( 1.5 ) # simulate doing work
                # HERE
                # HERE
                # HERE
            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()

동작화면

동작 설명

threaded버튼을 누르면
self.threadedButton 버튼의 command=self.onThreadedClicked 를 수행 하게되며,
onThreadedClicked() 함수가 실행하게 됩니다.
그러면, self.bgTask.start() 가 실행되고 bgTask는 self.bgTask = BackgroundTask( self.myLongProcess ) 코드에 의해 미리 등록되어 있으므로, myLongProcess 함수가 실행 됩니다.

myLongProcess 내에서는 긴작업을 할 수 있으나 여기에서는 같은 작업을 10회 하도록 코드가 되어있습니다. 그러면서 아래쪽 숫자가 증가 하게 됩니다.

아래 코드가 실제 긴작업을 할 수 있는 코드 입니다.

        def myLongProcess( self, isRunningFunc=None ) :
            print ("starting myLongProcess")
            # WORKING THREAD

긴작업 동안에 중단이 오는경우를 위해서 주기적으로 thread가 중지 요청이 되었는지 확인이 필요하기 때문에 아래와 같은 코드가 들어 있습니다.

                    if not isRunningFunc() :
                        self.onMyLongProcessUpdate( "Stopped!" )