SoFunction
Updated on 2025-03-10

Detailed explanation of several examples of asynchronous tasks multi-threading

Multi-threaded asynchronous threads are commonly used by us, such as when we are performing time-consuming operations and do not want to use the main program;

1. QThread 

from  import QThread, pyqtSignal
from  import QApplication, QLabel, QVBoxLayout, QWidget, QPushButton
import time
class WorkerThread(QThread):
    progress = pyqtSignal(int)  # Define the signal    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()
            (i)  # Send a signalclass MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        (800, 600)
        ()
    def initUI(self):
         = QLabel("Progress: 0")
         = QPushButton("Start the mission")
        (self.start_task)
        layout = QVBoxLayout()
        ()
        ()
        (layout)
    def excuteSomeThing(self):
        (0.1)  # Simulation time-consuming operation    def start_task(self):
         = WorkerThread(self)
        (self.update_label)  # Connect signal to slot function        ()  # Start the thread    def update_label(self, value):
        (f"schedule: {value}")
app = QApplication([])
window = MainWindow()
()
app.exec_()

The main thread function is executed in the child thread, and it is in the child thread;

2. QThreadPool

from  import QRunnable, QThreadPool, pyqtSignal, QObject
from  import QApplication, QLabel, QVBoxLayout, QWidget, QPushButton
import time
class WorkerSignals(QObject):
    progress = pyqtSignal(int)
class Worker(QRunnable):
    def __init__(self):
        super().__init__()
         = WorkerSignals()
    def run(self):
        for i in range(1, 101):
            (0.01)  # Simulation time-consuming operation            (i)  # Send a signalclass MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        (800, 600)
        ()
        self.thread_pool = QThreadPool()
    def initUI(self):
         = QLabel("Progress: 0")
         = QPushButton("Start the mission")
        (self.start_task)
        layout = QVBoxLayout()
        ()
        ()
        (layout)
    def start_task(self):
        worker = Worker()
        (self.update_label)
        self.thread_pool.start(worker)
    def update_label(self, value):
        (f"schedule: {value}")
app = QApplication([])
window = MainWindow()
()
app.exec_()

from  import pyqtSignal, QObject
from  import QApplication, QLabel, QVBoxLayout, QWidget, QPushButton
from  import ThreadPoolExecutor
import time
class Worker(QObject):
    progress = pyqtSignal(int)
    def do_work(self):
        for i in range(1, 101):
            (0.021)  # Simulation time-consuming operation            (i)
class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        (800, 600)
        ()
         = ThreadPoolExecutor(max_workers=10)
    def initUI(self):
         = QLabel("Progress: 0")
         = QPushButton("Start the mission")
        (self.start_task)
        layout = QVBoxLayout()
        ()
        ()
        (layout)
    def start_task(self):
         = Worker()
        (self.update_label)
        (.do_work)
    def update_label(self, value):
        (f"schedule: {value}")
app = QApplication([])
window = MainWindow()
()
app.exec_()

Summarize

  • QThread: Suitable for scenarios where custom thread logic is required.
  • QRunnable + QThreadPool: Suitable for lightweight and high-concurrency tasks.
  • : Simple combination of signal and slot mechanism to use thread pool.

This is the end of this article about several examples of PyQt asynchronous tasks multi-threading. For more related PyQt multi-threading content, please search for my previous articles or continue browsing the related articles below. I hope you will support me more in the future!