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 = 0layout = 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 += 1self.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 = 1000for n in range(total_n):progress_pc = int(100 * float(n + 1) / total_n) #Progress 0-100% as intself.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)# Executeself.threadpool.start(worker)# 接收progress信号，并显示def update_progress(self, progress_value):self.progressbar.setValue(progress_value)