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)