PyQt 多线程

1 卡住的计时器

我们定义了一个计时器，每秒钟更新一次显示的数字。此外我们定义了一个耗时5秒的任务oh_no，和按钮“危险”绑定。

当我们点击“危险”按钮时，程序去执行oh_no，导致显示停止更新了。

import sys
import time
from PyQt6.QtCore import QTimer
from PyQt6.QtWidgets import (
    QApplication,
    QLabel,
    QMainWindow,
    QPushButton,
    QVBoxLayout,
    QWidget,
)

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        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()

        # 定时器，每1秒更新一次文本
        self.timer = QTimer()
        self.timer.setInterval(1000)
        self.timer.timeout.connect(self.recurring_timer)
        self.timer.start()

    def oh_no(self):
        time.sleep(5)
    def recurring_timer(self):
        self.counter += 1
        self.l.setText("Counter: %d" % self.counter)
	
app = QApplication(sys.argv)
window = MainWindow()
app.exec()

QT提供了线程的接口，主要通过两个类实现
QRunnable: 工作的容器
QThreadPool：线程池

继承QRunnable并实现run方法：

class Worker(QRunnable):
	"""
	Worker thread
	"""
	@pyqtSlot()
	def run(self):
		"""
		Your code goes in this function
		"""
		print("Thread start")
		time.sleep(5)
		print("Thread complete")

创建线程池：

class MainWindow(QMainWindow):
	def __init__(self):
		super().__init__()
		self.threadpool = QThreadPool()
		print(
		"Multithreading with maximum %d threads" % self.
		threadpool.maxThreadCount()
		)

使用线程池启动任务：

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

使用线程后，当我们点击危险时会启动额外的线程去执行任务，不会阻塞Qt的显示。

2 进度条

当我们执行一个耗时的任务时，常见的做法是添加一个进度条来让用户了解任务的进度。

为此，我们需要在任务中发送进度信息，然后在Qt窗口中更新进度。

1.导入相关库

import sys
import time
from PyQt6.QtCore import QObject, QRunnable, QThreadPool, QTimer,\
                         pyqtSignal, pyqtSlot
from PyQt6.QtWidgets import (
    QApplication,
    QLabel,
    QMainWindow,
    QProgressBar,
    QPushButton,
    QVBoxLayout,
    QWidget,
)

2.在任务中使用信号量发送进度

# 信号量，用于表示进度
class WorkerSignals(QObject):
    progress = pyqtSignal(int)

class Worker(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = WorkerSignals()
        
    @pyqtSlot()
    def run(self):
        total_n = 1000
        for n in range(total_n):
            progress_pc = int(100 * float(n + 1) / total_n) #Progress 0-100% as int
            self.signals.progress.emit(progress_pc) # 通过信号发送当前进度值
            time.sleep(0.01)  

3.在窗口中接收信号，并在进度条中显示

class MainWindow(QMainWindow):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        layout = QVBoxLayout()
        self.progressbar = QProgressBar() # 进度条
        button = QPushButton("启动")
        button.pressed.connect(self.execute)

        layout.addWidget(self.progressbar)
        layout.addWidget(button)
        w = QWidget()
        w.setLayout(layout)
        self.setCentralWidget(w)
        self.show()

        self.threadpool = QThreadPool()
        print(
            "Multithreading with maximum %d threads" % self.
            threadpool.maxThreadCount()
        )
    def execute(self):
        worker = Worker()
        # 和update_progress连接，
        worker.signals.progress.connect(self.update_progress)
        # Execute
        self.threadpool.start(worker)
    # 接收progress信号，并显示
    def update_progress(self, progress_value):
        self.progressbar.setValue(progress_value)