Python의 PyQt 라이브러리를 활용하여 스레드를 관리하는 방법

PyQt는 Python 애플리케이션에서 그래픽 사용자 인터페이스를 구축하기 위한 강력한 프레임워크를 제공합니다. 백그라운드 작업을 실행하는 동안 원활하고 응답성이 뛰어난 UI 업데이트를 보장하기 위해 Qt는 기본 GUI 스레드와 동시에 실행되는 스레드를 활용합니다. 그러나 스레드를 효과적으로 사용하려면 신중한 고려가 필요합니다.

Maya Posch의 기사에 설명된 대로 run 메소드의 재구현은 QThreads로 작업할 때 선호되는 접근 방식이 되어서는 안 됩니다. 대신, 스레드 간의 통신을 용이하게 하기 위해 신호와 슬롯을 사용하는 것을 고려하십시오. 올바른 구현을 설명하기 위해 실제적인 예가 제시됩니다.

PyQt 스레드 관리의 예

샘플에서는 긴 계산을 처리하는 별도의 작업자 스레드를 생성합니다. 메인 GUI 스레드는 그래픽 인터페이스를 관리합니다. 작업자 스레드는 신호를 통해 GUI에 상태 업데이트를 전달합니다.

계산을 시작하려면 사용자가 "시작" 버튼을 클릭합니다. "취소" 버튼을 사용하여 작업을 종료하고 작업자 스레드를 재설정할 수 있습니다. 강제 스레드 종료는 일반적으로 권장되지 않지만 데모 목적으로 사용됩니다.

Python Code

from PyQt4 import QtGui, QtCore
import sys
import random

class Example(QtCore.QObject):

    signalStatus = QtCore.pyqtSignal(str)

    def __init__(self, parent=None):
        super(self.__class__, self).__init__(parent)

        # Create a gui object.
        self.gui = Window()

        # Create a new worker thread.
        self.createWorkerThread()

        # Make any cross object connections.
        self._connectSignals()

        self.gui.show()


    def _connectSignals(self):
        self.gui.button_cancel.clicked.connect(self.forceWorkerReset)
        self.signalStatus.connect(self.gui.updateStatus)
        self.parent().aboutToQuit.connect(self.forceWorkerQuit)


    def createWorkerThread(self):

        # Setup the worker object and the worker_thread.
        self.worker = WorkerObject()
        self.worker_thread = QtCore.QThread()
        self.worker.moveToThread(self.worker_thread)
        self.worker_thread.start()

        # Connect any worker signals
        self.worker.signalStatus.connect(self.gui.updateStatus)
        self.gui.button_start.clicked.connect(self.worker.startWork)


    def forceWorkerReset(self):      
        if self.worker_thread.isRunning():
            print('Terminating thread.')
            self.worker_thread.terminate()

            print('Waiting for thread termination.')
            self.worker_thread.wait()

            self.signalStatus.emit('Idle.')

            print('building new working object.')
            self.createWorkerThread()


    def forceWorkerQuit(self):
        if self.worker_thread.isRunning():
            self.worker_thread.terminate()
            self.worker_thread.wait()


class WorkerObject(QtCore.QObject):

    signalStatus = QtCore.pyqtSignal(str)

    def __init__(self, parent=None):
        super(self.__class__, self).__init__(parent)

    @QtCore.pyqtSlot()        
    def startWork(self):
        for ii in range(7):
            number = random.randint(0,5000**ii)
            self.signalStatus.emit('Iteration: {}, Factoring: {}'.format(ii, number))
            factors = self.primeFactors(number)
            print('Number: ', number, 'Factors: ', factors)
        self.signalStatus.emit('Idle.')

    def primeFactors(self, n):
        i = 2
        factors = []
        while i * i  1:
            factors.append(n)
        return factors


class Window(QtGui.QWidget):

    def __init__(self):
        QtGui.QWidget.__init__(self)
        self.button_start = QtGui.QPushButton('Start', self)
        self.button_cancel = QtGui.QPushButton('Cancel', self)
        self.label_status = QtGui.QLabel('', self)

        layout = QtGui.QVBoxLayout(self)
        layout.addWidget(self.button_start)
        layout.addWidget(self.button_cancel)
        layout.addWidget(self.label_status)

        self.setFixedSize(400, 200)

    @QtCore.pyqtSlot(str)
    def updateStatus(self, status):
        self.label_status.setText(status)


if __name__=='__main__':
    app = QtGui.QApplication(sys.argv)
    example = Example(app)
    sys.exit(app.exec_())

이 예는 PyQt 애플리케이션에서 QThreads의 적절한 사용을 보여주며 사용자 인터페이스를 정지시키지 않고 효율적인 백그라운드 작업을 허용합니다.