Python界面(GUI)编程PyQt5之多线程应用程序

背景

基于Qt的应用程序(像大多数GUI应用程序一样)是基于事件的。这意味着执行是根据用户交互、信号和计时器来驱动的。在事件驱动的应用程序中，单击按钮将创建一个事件，应用程序随后将处理该事件以产生预期的输出。事件被推入和从事件队列取出，然后按顺序处理。

事件循环通过在QApplication对象上调用.exec_()开始，并在与Python代码相同的线程中运行。运行这个事件循环的线程——通常称为GUI线程——也处理与主机操作系统的所有窗口通信。

默认情况下，由事件循环触发的任何执行也将在这个线程中同步运行。在实践中，这意味着当PyQt应用程序在代码中执行某些操作时，窗口通信和GUI交互将被阻塞。

如果您所做的很简单，并且快速地将控制返回到GUI循环，那么用户将无法察觉这种阻塞。但是，如果您需要执行长时间运行的任务，例如打开/写入一个大文件、下载一些数据或渲染一些复杂的界面，那么就会出现问题。对用户来说，应用程序似乎没有响应(因为它确实没有响应)。因为你的应用程序不再与操作系统通信，在MacOS X上，如果你点击你的应用程序，你会看到旋转的死亡轮。没有人希望这样。

解决方案很简单:将您的工作从GUI线程中取出(放到另一个线程中)。PyQt(通过Qt)提供了一个简单的接口来实现这一点。

线程和进程

在PyQt应用程序中运行独立任务有两种主要方法:线程和进程。

线程共享相同的内存空间，因此启动速度快，消耗的资源最少。共享内存使得在线程之间传递数据变得非常简单，但是从不同的线程读/写内存可能会导致竞争条件或分段错误。在Python中还有一个额外的问题，即多个线程被同一个全局解释器锁(GIL)绑定——这意味着非GIL释放的Python代码一次只能在一个线程中执行。但是，这不是PyQt的主要问题，因为大部分时间都花在Python之外。

进程使用独立的内存空间(和一个完全独立的Python解释器)。这样可以避免GIL的任何潜在问题，但代价是启动时间较慢、内存开销较大以及发送/接收数据的复杂性。

为了简单起见，这篇文章将使用线程演示；

QRunnable和QThreadPool

Qt为在线程中运行提供了非常简单的接口。它是围绕两个类构建的:QRunnable和QThreadPool。前者是您想要执行的工作的容器，而后者是您将该工作传递给其他线程的方法。

from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *

import time

class MainWindow(QWidget):

    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)
        self.setWindowTitle("高效码农")
        
        self.threadpool = QThreadPool()

        Vlayout = QVBoxLayout()
        okButton = QPushButton("OK")
        okButton.clicked.connect(self.oh_no)
        

        Vlayout.addWidget(okButton)

        self.setLayout(Vlayout)
        self.show()

    def oh_no(self):
        worker = Worker()
        self.threadpool.start(worker) 


class Worker(QRunnable):

    @pyqtSlot()
    def run(self):
        print("Thread start") 
        time.sleep(5)
        print("Thread complete")


app = QApplication([])
window = MainWindow()
app.exec_()

创建一个Worker实例,然后传递给self.threadpool；现在，单击该按钮将创建一个工作程序来处理（长时间运行）进程，并通过线程池将其拆分为另一个线程。

改进QRunnables

如果要将自定义数据传递给执行功能，可以通过init进行操作，然后可以用self.从run插槽中访问数据。

class Worker(QRunnable):
    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()
        # Store constructor arguments (re-used for processing)
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

    @pyqtSlot()
    def run(self):
        self.fn(*self.args, **self.kwargs)

线程IO

有时候，能够从运行的线程回调状态和数据是非常重要的。这可能包括计算结果、引发的异常或正在进行的进程(考虑进度条)。Qt提供的signals and slot框架允许您这样做，并且是线程安全的，允许从运行的线程直接安全通信到GUI前端。信号允许您发出.emit值，然后通过与.connect链接的插槽函数在代码的其他地方接收这些值。

下面是一个简单的WorkerSignals类，它定义为包含许多示例信号。

from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *

import time
import traceback, sys


class WorkerSignals(QObject):
    '''
    Defines the signals available from a running worker thread.

    Supported signals are:

    finished
        No data
    
    error
        `tuple` (exctype, value, traceback.format_exc() )
    
    result
        `object` data returned from processing, anything

    progress
        `int` indicating % progress 

    '''
    finished = pyqtSignal()
    error = pyqtSignal(tuple)
    result = pyqtSignal(object)
    progress = pyqtSignal(int)


class Worker(QRunnable):
    '''
    Worker thread

    Inherits from QRunnable to handler worker thread setup, signals and wrap-up.

    :param callback: The function callback to run on this worker thread. Supplied args and 
                     kwargs will be passed through to the runner.
    :type callback: function
    :param args: Arguments to pass to the callback function
    :param kwargs: Keywords to pass to the callback function

    '''

    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()

        # Store constructor arguments (re-used for processing)
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()    

        # Add the callback to our kwargs
        self.kwargs['progress_callback'] = self.signals.progress        

    @pyqtSlot()
    def run(self):
        '''
        Initialise the runner function with passed args, kwargs.
        '''
        
        # Retrieve args/kwargs here; and fire processing using them
        try:
            result = self.fn(*self.args, **self.kwargs)
        except:
            traceback.print_exc()
            exctype, value = sys.exc_info()[:2]
            self.signals.error.emit((exctype, value, traceback.format_exc()))
        else:
            self.signals.result.emit(result)  # Return the result of the processing
        finally:
            self.signals.finished.emit()  # Done
        


class MainWindow(QMainWindow):


    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)
    
        self.counter = 0
    
        layout = QVBoxLayout()
        
        self.l = QLabel("Start")
        b = QPushButton("DANGER!")
        b.pressed.connect(self.oh_no)
    
        layout.addWidget(self.l)
        layout.addWidget(b)
    
        w = QWidget()
        w.setLayout(layout)
    
        self.setCentralWidget(w)
    
        self.show()

        self.threadpool = QThreadPool()
        print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())

        self.timer = QTimer()
        self.timer.setInterval(1000)
        self.timer.timeout.connect(self.recurring_timer)
        self.timer.start()
    
    def progress_fn(self, n):
        print("%d%% done" % n)

    def execute_this_fn(self, progress_callback):
        for n in range(0, 5):
            time.sleep(1)
            progress_callback.emit(n*100/4)
            
        return "Done."
 
    def print_output(self, s):
        print(s)
        
    def thread_complete(self):
        print("THREAD COMPLETE!")
 
    def oh_no(self):
        # Pass the function to execute
        worker = Worker(self.execute_this_fn) # Any other args, kwargs are passed to the run function
        worker.signals.result.connect(self.print_output)
        worker.signals.finished.connect(self.thread_complete)
        worker.signals.progress.connect(self.progress_fn)
        
        # Execute
        self.threadpool.start(worker) 

        
    def recurring_timer(self):
        self.counter +=1
        self.l.setText("Counter: %d" % self.counter)
    
    
app = QApplication([])
window = MainWindow()
app.exec_()