Python - PyQt5에서 Threadpool 사용하는 방법

pythontips

PyQt5에서 Threadpool을 사용하려고 했는데, 익숙하지 않아 시행착오가 좀 있었습니다.

다행히 좋은 예제를 찾아서, 조금 수정하여 필요한 내용을 구현할 수 있었습니다.

이 글에서 간단히 소개합니다.

1. PyQt5의 Threadpool 예제

예제 코드는 아래와 같고, 출처는 GitHub입니다. 코드를 실행해보면 8개의 Thread pool이 생성되고, 버튼을 누르면 1개의 Task가 수행됩니다. 여러 작업이 동시에 수행되는 것 같지는 않습니다. 예제는 Threadpool을 Task를 처리했을 때와, UI Thread에서 처리했을 때의 차이점에 포커스를 맞춘 것 같습니다.

from PyQt5 import QtCore
from PyQt5.QtWidgets import QMainWindow, QApplication, QLabel, QPushButton, QVBoxLayout, QWidget
from PyQt5.QtCore import Qt, QObject, QRunnable, pyqtSlot, QThreadPool, QTimer
import traceback, sys
import datetime

# this may be a long sample but you can copy and pase it for test
# the different parts of code are commented so it can be clear what it does

class WorkerSignals(QObject):
    finished = QtCore.pyqtSignal() # create a signal
    result = QtCore.pyqtSignal(object) # create a signal that gets an object as argument

class Worker(QRunnable):
    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()
        self.fn = fn # Get the function passed in
        self.args = args # Get the arguments passed in
        self.kwargs = kwargs # Get the keyward arguments passed in
        self.signals = WorkerSignals() # Create a signal class

    @pyqtSlot()
    def run(self): # our thread's worker function
        result = self.fn(*self.args, **self.kwargs) # execute the passed in function with its arguments
        self.signals.result.emit(result)  # return result
        self.signals.finished.emit()  # emit when thread ended

class MainWindow(QMainWindow):
    def __init__(self, *args, **kwargs): #-----------------------------------------
        super(MainWindow, self).__init__(*args, **kwargs) #                        |
        self.threadpool = QThreadPool() #                                          |
        print("Maximum Threads : %d" % self.threadpool.maxThreadCount()) #         |
        self.layout = QVBoxLayout() #                                              |
        self.time_label = QLabel("Start") #                                        |
        self.btn_thread = QPushButton("Using Thread") #                            |
        self.btn_thread.pressed.connect(self.threadRunner) #                       |
        self.btn_window = QPushButton("Without Thread") #                          |----- These are just some initialization
        self.layout.addWidget(self.time_label) #                                   |      
        self.layout.addWidget(self.btn_thread) #                                   |
        self.layout.addWidget(self.btn_window) #                                   |
        w = QWidget() #                                                            |
        w.setLayout(self.layout) #                                                 |
        self.setCentralWidget(w) #                                                 |
        self.show() #                                                              |
        self.timer = QTimer() #                                                    |
        self.timer.setInterval(10) #                                               |
        self.timer.timeout.connect(self.time) #                                    |
        self.timer.start() #-------------------------------------------------------

        # This button will run foo_window function in window ( main thread ) which will freeze our program
        self.btn_window.pressed.connect(lambda : self.foo_window(1))

        # a function that we'll use in our thread which doesn't make our program freeze
        # it will take around 5 seconds to process
    def foo_thread(self, num):
        # some long processing
        self.btn_thread.setText("Processing...")
        for i in range(50000000):
            num += 10
        return num

        # we'll use this function in window ( window = main thread )
        # it will take around 5 seconds to process
    def foo_window(self, num):
        # some long processing
        print("Window Processing...")
        for i in range(50000000):
            num += 10
        # add a label to layout
        label = QLabel("Result from Window is : "+str(num))
        self.layout.addWidget(label)
        return num

        # we'll use this function when 'finished' signal is emited
    def thread_finished(self):
        print("Finished signal emited.")
        self.btn_thread.setText("Using Thread")

        # we'll use this function when 'result' signal is emited
    def thread_result(self, s):
        # add a label to layout
        label = QLabel("Result from Thread is : "+str(s))
        self.layout.addWidget(label) # add a new label to window with the returned result from our thread

        # in this function we create our thread and run it
    def threadRunner(self):
        worker = Worker(self.foo_thread, num=1) # create our thread and give it a function as argument with its args
        worker.signals.result.connect(self.thread_result) # connect result signal of our thread to thread_result
        worker.signals.finished.connect(self.thread_finished) # connect finish signal of our thread to thread_complete
        self.threadpool.start(worker) # start thread

        # this function just gets current time and displays it in Window
    def time(self):
        now = datetime.datetime.now().time() # current time
        self.time_label.setText("Current Time: "+ str(now)) # desplay current time

if __name__ == "__main__":
    app = QApplication([])
    window = MainWindow()
    app.exec_()

2. Threadpool 예제 코드 분석

중요 코드만 간단히 소개하겠습니다.

2.1 Worker, WorkerSignals

Worker는 Task를 표현하는 단위입니다. Thread에 어떤 작업을 수행하려면 Worker를 만들고 Threadpool에서 실행시켜야 합니다. WorkerSignals은 Thread에서 처리되는 작업의 결과나 완료되었을 때, MainWindow로 이벤트를 콜백하는데 사용되는 객체입니다.

class WorkerSignals(QObject):
    finished = QtCore.pyqtSignal() # create a signal
    result = QtCore.pyqtSignal(object) # create a signal that gets an object as argument

class Worker(QRunnable):
    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()
        self.fn = fn # Get the function passed in
        self.args = args # Get the arguments passed in
        self.kwargs = kwargs # Get the keyward arguments passed in
        self.signals = WorkerSignals() # Create a signal class

    @pyqtSlot()
    def run(self): # our thread's worker function
        result = self.fn(*self.args, **self.kwargs) # execute the passed in function with its arguments
        self.signals.result.emit(result)  # return result
        self.signals.finished.emit()  # emit when thread ended

2.2 Task 생성 및 실행

어떤 작업을 수행할 때, Worker를 만들고 인자로 쓰레드에서 실행될 function을 전달합니다. worker.signals.result.connect()는 Task의 작업이 완료되었을 때 콜백되는 function을 설정하는 것이고, worker.signals.finished.connect()는 쓰레드가 종료될 때 콜백되는 function을 설정하는 것입니다.

worker = Worker(self.foo_thread, num=1) # create our thread and give it a function as argument with its args
worker.signals.result.connect(self.thread_result) # connect result signal of our thread to thread_result
worker.signals.finished.connect(self.thread_finished) # connect finish signal of our thread to thread_complete
self.threadpool.start(worker) # start thread

WorkerSignals를 보시면 finished와 result의 콜백이 전달할 인자 개수를 설정할 수 있습니다. finished는 인자가 없고, result는 1개의 인자를 전달합니다.

class WorkerSignals(QObject):
    finished = QtCore.pyqtSignal() # create a signal
    result = QtCore.pyqtSignal(object) # create a signal that gets an object as argument

Worker가 Threadpool로 전달되면, run(self)이 호출되며, Task 실행 및 그 결과를 MainWindow로 콜백합니다.

class Worker(QRunnable):
    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()
        self.fn = fn # Get the function passed in
        self.args = args # Get the arguments passed in
        self.kwargs = kwargs # Get the keyward arguments passed in
        self.signals = WorkerSignals() # Create a signal class

    @pyqtSlot()
    def run(self): # our thread's worker function
        result = self.fn(*self.args, **self.kwargs) # execute the passed in function with its arguments
        self.signals.result.emit(result)  # return result
        self.signals.finished.emit()  # emit when thread ended

3. 실행 모습

예제를 실행시키고 버튼을 누르면 다음과 같이 동작합니다.

pyqt5 threadpool

References

Loading script...

Related Posts

codechachaCopyright ©2019 codechacha