PyQt 異步任務多線程的幾種方案示例詳解

更新時間：2025年01月22日 09:40:05 作者：戀戀西風

文章總結了多線程異步編程的不同方式,包括QThread、QThreadPool和concurrent.futures,分別適用于不同場景,本文結合實例代碼給大家介紹的非常詳細,感興趣的朋友一起看看吧

多線程異步線程是我們常用的，如我們在執(zhí)行耗時操作，又不想卡用主程序；

1. QThread

from PyQt5.QtCore import QThread, pyqtSignal
from PyQt5.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget, QPushButton
import time
class WorkerThread(QThread):
    progress = pyqtSignal(int)  # 定義信號
    def __init__(self,main_instance):
        QThread.__init__(self)
        self.main_instance = main_instance
    def run(self):
        for i in range(1, 101):
            self.main_instance.excuteSomeThing()
            self.progress.emit(i)  # 發(fā)送信號
class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.resize(800, 600)
        self.initUI()
    def initUI(self):
        self.label = QLabel("進度: 0")
        self.button = QPushButton("開始任務")
        self.button.clicked.connect(self.start_task)
        layout = QVBoxLayout()
        layout.addWidget(self.label)
        layout.addWidget(self.button)
        self.setLayout(layout)
    def excuteSomeThing(self):
        time.sleep(0.1)  # 模擬耗時操作
    def start_task(self):
        self.worker = WorkerThread(self)
        self.worker.progress.connect(self.update_label)  # 連接信號到槽函數
        self.worker.start()  # 啟動線程
    def update_label(self, value):
        self.label.setText(f"進度: {value}")
app = QApplication([])
window = MainWindow()
window.show()
app.exec_()

子線程中回調主線程函數執(zhí)行，在子線程；

2. QThreadPool

from PyQt5.QtCore import QRunnable, QThreadPool, pyqtSignal, QObject
from PyQt5.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget, QPushButton
import time
class WorkerSignals(QObject):
    progress = pyqtSignal(int)
class Worker(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = WorkerSignals()
    def run(self):
        for i in range(1, 101):
            time.sleep(0.01)  # 模擬耗時操作
            self.signals.progress.emit(i)  # 發(fā)送信號
class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.resize(800, 600)
        self.initUI()
        self.thread_pool = QThreadPool()
    def initUI(self):
        self.label = QLabel("進度: 0")
        self.button = QPushButton("開始任務")
        self.button.clicked.connect(self.start_task)
        layout = QVBoxLayout()
        layout.addWidget(self.label)
        layout.addWidget(self.button)
        self.setLayout(layout)
    def start_task(self):
        worker = Worker()
        worker.signals.progress.connect(self.update_label)
        self.thread_pool.start(worker)
    def update_label(self, value):
        self.label.setText(f"進度: {value}")
app = QApplication([])
window = MainWindow()
window.show()
app.exec_()

3.concurrent

from PyQt5.QtCore import pyqtSignal, QObject
from PyQt5.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget, QPushButton
from concurrent.futures import ThreadPoolExecutor
import time
class Worker(QObject):
    progress = pyqtSignal(int)
    def do_work(self):
        for i in range(1, 101):
            time.sleep(0.021)  # 模擬耗時操作
            self.progress.emit(i)
class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.resize(800, 600)
        self.initUI()
        self.executor = ThreadPoolExecutor(max_workers=10)
    def initUI(self):
        self.label = QLabel("進度: 0")
        self.button = QPushButton("開始任務")
        self.button.clicked.connect(self.start_task)
        layout = QVBoxLayout()
        layout.addWidget(self.label)
        layout.addWidget(self.button)
        self.setLayout(layout)
    def start_task(self):
        self.worker = Worker()
        self.worker.progress.connect(self.update_label)
        self.executor.submit(self.worker.do_work)
    def update_label(self, value):
        self.label.setText(f"進度: {value}")
app = QApplication([])
window = MainWindow()
window.show()
app.exec_()

總結