PYQT TUTORIAL
Hướng dẫn sử dụng LangChain bằng Python Cách tự viết Context Manager trong Python Cách background của hình ảnh trong Python Cách làm việc với Notion API trong Python Cách đo Time chạy trong Python Cách sao chép danh sách (List) trong Python Cách kiểm tra List trống trong Python Cách sắp xếp dictionary theo giá trị trong Python PyQt là gì? Tạo một chương trình "Hello World". Tín hiệu và Khe (Signals & Slots) trong PyQt Cách sử dụng widget PyQt QLabel Cách sử dụng widget QPushButton của PyQt Cách sử dụng widget QLineEdit trong PyQt Cách sử dụng QVBoxLayout trong PyQt Cách sử dụng QGridLayout trong PyQt Cách sử dụng QFormLayout trong PyQt Cách sử dụng lớp QCheckBox trong PyQt Cách sử dụng lớp PyQt QRadioButton Sử dụng PyQt QComboBox để tạo Widget Combobox Cách sử dụng widget PyQt QSpinBox để tạo một spin box Cách tạo một widget nhập ngày sử dụng lớp PyQt QDateEdit Cách tạo một widget nhập giờ sử dụng lớp PyQt QTimeEdit Cách tạo một widget nhập ngày và giờ sử dụng PyQt QDateTimeEdit Cách sử dụng lớp PyQt QSlider để tạo một widget thanh trượt (slider). Cách dùng PyQt QWidget để làm container chứa các widget khác. Cách dùng lớp PyQt QTabWidget để tạo một widget dạng tab Cách dùng lớp PyQt QGroupBox để tạo một khung nhóm với tiêu đề Cách dùng lớp PyQt QTextEdit để tạo một widget cho phép chỉnh sửa Cách sử dụng lớp PyQt QProgressBar để tạo một widget progress bar Cách sử dụng lớp PyQt QMessageBox để tạo một hộp thoại Cách dùng lớp PyQt QInputDialog để tạo một hộp thoại nhập liệu Cách dùng lớp PyQt QFileDialog để tạo hộp thoại chọn file Cách sử dụng lớp QMainWindow của PyQt để tạo cửa sổ Cách sử dụng lớp PyQt QMenu để tạo menu Cách dùng lớp PyQt QToolBar để tạo các widget toolbar Cách dùng lớp QDockWidget của PyQt để tạo một widget Cách dùng lớp QStatusBar trong PyQt để tạo thanh status bar Cách sử dụng lớp QListWidget trong Python Cách sử dụng lớp QTableWidget để tạo một bảng Cách sử dụng lớp QTreeWidget của PyQt Cách sử dụng Qt Designer để thiết kế UX/UI trong PyQt. Cách sử dụng QThread trong PyQt Cách tạo các lớp QThreadPool đa luồng trong PyQt Cách hoạt động của Model/View trong PyQt Cách sử dụng Qt Style Sheets (QSS) trong PyQt. Chuyển đổi PyQt sang EXE trong PyQt
CÁC CHỦ ĐỀ
BÀI MỚI NHẤT
MỚI CẬP NHẬT

Thông báo: Download 4 khóa học Python từ cơ bản đến nâng cao tại đây.

Cách tạo các lớp QThreadPool đa luồng trong PyQt

PyQt, một thư viện mạnh mẽ dành cho việc phát triển ứng dụng giao diện đồ họa trên nền tảng Python, cung cấp các công cụ tuyệt vời để xử lý đa luồng. Trong số đó, các lớp QThreadPoolQRunnable đóng vai trò then chốt trong việc quản lý và thực thi các tác vụ đồng thời.

test php

banquyen png
Bài viết này được đăng tại freetuts.net, không được copy dưới mọi hình thức.

Hướng dẫn này sẽ đưa bạn qua các bước để tạo ra một ứng dụng PyQt hiệu quả sử dụng các lớp QThreadPoolQRunnable. Bạn sẽ học cách tối ưu hóa ứng dụng của mình bằng cách chạy các tác vụ nền mà không làm ảnh hưởng đến giao diện người dùng chính. Mình sẽ bắt đầu với các khái niệm cơ bản và dần dần đi sâu vào cách sử dụng các lớp này để xây dựng các ứng dụng PyQt mạnh mẽ và đáp ứng nhanh.

Giới thiệu về các lớp QThreadPool & QRunnable trong PyQt

Lớp QThread cho phép bạn chuyển một tác vụ dài hạn sang một luồng công việc để làm cho ứng dụng trở nên phản hồi nhanh hơn. Lớp QThread hoạt động tốt nếu ứng dụng của bạn có một số ít luồng công việc.

Một chương trình đa luồng sẽ hiệu quả hơn khi số lượng đối tượng QThread tương ứng với số lõi CPU.

Bài viết này được đăng tại [free tuts .net]

Ngoài ra, việc tạo ra các luồng tiêu tốn tài nguyên máy tính khá lớn. Do đó, chương trình nên tái sử dụng các luồng đã tạo ra càng nhiều càng tốt.

Vì vậy, việc sử dụng lớp QThread để quản lý các luồng công việc gặp phải hai thách thức chính:

  • Xác định số lượng luồng lý tưởng cho ứng dụng dựa trên số lõi CPU.
  • Tái sử dụng và tái chế các luồng càng nhiều càng tốt.

May mắn thay, PyQt cung cấp lớp QThreadPool để giải quyết những thách thức này cho bạn. Lớp QThreadPool thường được sử dụng cùng với lớp QRunnable.

Lớp QRunnable đại diện cho một tác vụ mà bạn muốn thực thi trong một luồng công việc. Lớp QThreadPool thực thi một đối tượng QRunnable và tự động quản lý và tái chế các luồng.

Mỗi ứng dụng Qt có một đối tượng QThreadPool toàn cục, có thể được truy cập thông qua phương thức tĩnh globalInstance() của lớp QThreadPool.

Để sử dụng các lớp QThreadPool và QRunnable, bạn thực hiện theo các bước sau:

Đầu tiên, tạo một lớp kế thừa từ lớp QRunnable và ghi đè phương thức run():

class Worker(QRunnable):
    @Slot()
    def run(self):
        # thực hiện tác vụ dài hạn ở đây
        pass

Thứ hai, truy cập vào hồ bơi luồng từ cửa sổ chính và khởi chạy các luồng công việc:

class MainWindow(QMainWindow):
    # các phương thức khác
    # ...

    def start(self):
        """ Tạo và thực thi các luồng công việc """
        pool = QThreadPool.globalInstance()
        for _ in range(1, 100):
            pool.start(Worker())

Để cập nhật tiến độ của công việc từ luồng công việc về luồng chính, bạn sử dụng tín hiệu và slot. Tuy nhiên, lớp QRunnable không hỗ trợ tín hiệu.

Do đó, bạn cần định nghĩa một lớp riêng kế thừa từ lớp QObject và sử dụng lớp đó trong lớp Worker. Các bước thực hiện như sau:

Đầu tiên, định nghĩa lớp Signals kế thừa từ lớp QObject:

class Signals(QObject):
    completed = Signal()

Trong lớp Signals, chúng ta định nghĩa một tín hiệu gọi là completed. Lưu ý rằng bạn có thể định nghĩa nhiều tín hiệu theo nhu cầu.

Thứ hai, phát tín hiệu completed khi công việc hoàn tất trong lớp Worker:

class Runnable(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = Signals()

    @Slot()
    def run(self):
        # tác vụ dài hạn
        # ...
        # phát tín hiệu completed
        self.signals.completed.emit()

Thứ ba, kết nối tín hiệu của luồng công việc với slot của cửa sổ chính trước khi gửi công việc đến hồ bơi:

class MainWindow(QMainWindow):
    # các phương thức khác
    # ...

    def start(self):
        """ Tạo và thực thi các luồng công việc """
        pool = QThreadPool.globalInstance()
        for _ in range(1, 100):
            worker = Worker()
            worker.signals.completed.connect(self.update)
            pool.start(worker)

    def update(self):
        # cập nhật công việc
        pass

Ví dụ về QThreadPool trong PyQt

Dưới đây là ví dụ về cách sử dụng các lớp QThreadPool và QRunnable:

import sys
import time
from PyQt6.QtWidgets import QApplication, QMainWindow, QPushButton, QGridLayout, QWidget, QProgressBar, QListWidget
from PyQt6.QtCore import QRunnable, QObject, QThreadPool, pyqtSignal as Signal, pyqtSlot as Slot

class Signals(QObject):
    started = Signal(int)
    completed = Signal(int)

class Worker(QRunnable):
    def __init__(self, n):
        super().__init__()
        self.n = n
        self.signals = Signals()

    @Slot()
    def run(self):
        self.signals.started.emit(self.n)
        time.sleep(self.n*1.1)
        self.signals.completed.emit(self.n)

class MainWindow(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)

        self.setWindowTitle('QThreadPool Demo')

        self.job_count = 10
        self.comleted_jobs = []

        widget = QWidget()
        widget.setLayout(QGridLayout())
        self.setCentralWidget(widget)

        self.btn_start = QPushButton('Start', clicked=self.start_jobs)
        self.progress_bar = QProgressBar(minimum=0, maximum=self.job_count)
        self.list = QListWidget()

        widget.layout().addWidget(self.list, 0, 0, 1, 2)
        widget.layout().addWidget(self.progress_bar, 1, 0)
        widget.layout().addWidget(self.btn_start, 1, 1)

        self.show()

    def start_jobs(self):
        self.restart()
        pool = QThreadPool.globalInstance()
        for i in range(1, self.job_count+1):
            worker = Worker(i)
            worker.signals.completed.connect(self.complete)
            worker.signals.started.connect(self.start)
            pool.start(worker)

    def restart(self):
        self.progress_bar.setValue(0)
        self.comleted_jobs = []
        self.btn_start.setEnabled(False)

    def start(self, n):
        self.list.addItem(f'Job #{n} started...')

    def complete(self, n):
        self.list.addItem(f'Job #{n} completed.')
        self.comleted_jobs.append(n)
        self.progress_bar.setValue(len(self.comleted_jobs))

        if len(self.comleted_jobs) == self.job_count:
            self.btn_start.setEnabled(True)

if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = MainWindow()
    sys.exit(app.exec())

PyQt QThreadPool png

Lớp Signals

Mình định nghĩa lớp Signals kế thừa từ QObject. Lớp Signals có một biến lớp là completed, là một instance của Signal. Tín hiệu completed chứa một từ điển và được phát khi chương trình hoàn tất việc lấy giá cổ phiếu.

class Signals(QObject):
    completed = Signal(dict)

Lớp Stock

Lớp Stock kế thừa từ QRunnable. Nó ghi đè phương thức run() để lấy giá cổ phiếu từ trang web Yahoo Finance. Sau khi hoàn tất, phương thức run() phát tín hiệu completed với thông tin về cổ phiếu và giá của nó.

Nếu có lỗi xảy ra như không tìm thấy ký hiệu hoặc trang web thay đổi cách hiển thị giá cổ phiếu, phương thức run() trả về ký hiệu với giá là N/A.

class Stock(QRunnable):
    BASE_URL = 'https://finance.yahoo.com/quote/'

    def __init__(self, symbol):
        super().__init__()
        self.symbol = symbol
        self.signal = Signals()

    @Slot()
    def run(self):
        stock_url = f'{self.BASE_URL}{self.symbol}'
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"}
        response = requests.get(stock_url, headers=headers)
        if response.status_code != 200:
            self.signal.completed.emit({'symbol': self.symbol, 'price': 'N/A'})
            return

        tree = html.fromstring(response.text)
        price_text = tree.xpath(
            '//*[@id="quote-header-info"]/div[3]/div[1]/div[1]/fin-streamer[1]/text()'
        )

        if not price_text:
            self.signal.completed.emit({'symbol': self.symbol, 'price': 'N/A'})
            return

        price = float(price_text[0].replace(',', ''))

        self.signal.completed.emit({'symbol': self.symbol, 'price': price})

Lưu ý rằng Yahoo Finance có thể thay đổi cấu trúc của trang. Để chương trình hoạt động, bạn cần cập nhật XPath của giá cổ phiếu nếu cần.

Lớp MainWindow

Đầu tiên, đọc các ký hiệu từ một tệp và gán chúng cho biến self.symbols:

self.symbols = self.read_symbols(filename)

Phương thức read_symbols() trông như sau:

def read_symbols(self, filename):
    path = Path(filename)
    text = path.read_text()
    return [symbol.strip() for symbol in text.split('\n')]

File văn bản (symbols.txt) chứa mỗi ký hiệu trên một dòng:

AAPL	
MSFT
GOOG	
AMZN
TSLA
META
NVDA
BABA
CRM
INTC
PYPL	
AMD
ATVI
EA
TTD
ORCL

Tiếp theo, khởi tạo các widget và kết nối các tín hiệu với các slot:

self.start_button.clicked.connect(self.fetch_stock_prices)
self.results_list.setSortingEnabled(True)

Phương thức fetch_stock_prices() sẽ đọc các ký hiệu từ tệp, khởi tạo một đối tượng Stock cho mỗi ký hiệu, kết nối tín hiệu của mỗi đối tượng với một slot, và đưa chúng vào hồ bơi luồng.

def fetch_stock_prices(self):
    self.clear_results()
    self.progress_bar.setValue(0)
    pool = QThreadPool.globalInstance()
    for symbol in self.symbols:
        stock = Stock(symbol)
        stock.signal.completed.connect(self.update_stock_price)
        pool.start(stock)

Cuối cùng, xử lý kết quả từ tín hiệu completed trong phương thức update_stock_price():

def update_stock_price(self, result):
    symbol = result['symbol']
    price = result['price']
    self.results_list.addItem(f'{symbol}: ${price}')
    self.progress_bar.setValue(self.results_list.count())

Kết bài

Với hướng dẫn này, bạn đã được trang bị các kiến thức cần thiết để xây dựng các ứng dụng PyQt hiệu quả bằng cách tận dụng sức mạnh của đa luồng thông qua các lớp QThreadPoolQRunnable. Bằng cách áp dụng các kỹ thuật quản lý luồng và xử lý đồng thời, bạn có thể cải thiện đáng kể khả năng phản hồi của ứng dụng và xử lý nhiều tác vụ nền một cách mượt mà.

Việc sử dụng QThreadPool giúp bạn tối ưu hóa hiệu suất ứng dụng bằng cách tái sử dụng và quản lý các luồng một cách tự động, trong khi QRunnable cho phép bạn định nghĩa các tác vụ dài hạn một cách linh hoạt. Nhờ vào sự tách biệt rõ ràng giữa các tác vụ nền và giao diện người dùng, ứng dụng của bạn sẽ không bị gián đoạn và vẫn duy trì được sự tương tác mượt mà với người dùng.

Hy vọng rằng với những kiến thức và kỹ thuật này, bạn sẽ có thể áp dụng chúng vào các dự án của mình, nâng cao chất lượng và hiệu suất của các ứng dụng PyQt mà bạn phát triển. Chúc bạn thành công và sáng tạo trong việc xây dựng các ứng dụng đa luồng mạnh mẽ và hiệu quả!

Cùng chuyên mục:

Cách tạo thư mục lồng nhau (nested directory) trong Python

Cách tạo thư mục lồng nhau (nested directory) trong Python

Cách thêm số 0 vào đầu chuỗi trong Python

Cách thêm số 0 vào đầu chuỗi trong Python

Sự khác biệt giữa @classmethod, @staticmethod và instance methods trong Python

Sự khác biệt giữa @classmethod, @staticmethod và instance methods trong Python

Sự khác biệt giữa str và repr trong Python

Sự khác biệt giữa str và repr trong Python

Các cách nối hai danh sách trong Python

Các cách nối hai danh sách trong Python

Sự khác biệt giữa append() và extend() trong list Python

Sự khác biệt giữa append() và extend() trong list Python

5 lỗi thường gặp trong Python

5 lỗi thường gặp trong Python

Các tính năng mới trong Python 3.10

Các tính năng mới trong Python 3.10

Tạo app ghi chú trong Python với nhận dạng giọng nói và API Notion

Tạo app ghi chú trong Python với nhận dạng giọng nói và API Notion

Làm chủ Pattern Matching trong Python 3.10

Làm chủ Pattern Matching trong Python 3.10

Cách yêu cầu người dùng nhập liệu đến khi nhận được phản hồi hợp lệ trong Python

Cách yêu cầu người dùng nhập liệu đến khi nhận được phản hồi hợp lệ trong Python

8 Mẹo Refactor Code Python nhanh gọn (Phần 2)

8 Mẹo Refactor Code Python nhanh gọn (Phần 2)

Cách sao chép file trong Python

Cách sao chép file trong Python

31 Phương thức xử lý chuỗi (String) quan trọng trong Python

31 Phương thức xử lý chuỗi (String) quan trọng trong Python

Cách xóa file và thư mục trong Python

Cách xóa file và thư mục trong Python

Tìm hiểu về *args và **kwargs trong Python

Tìm hiểu về *args và **kwargs trong Python

Cách làm phẳng danh sách lồng nhau trong Python

Cách làm phẳng danh sách lồng nhau trong Python

Phân tích dữ liệu Apple Health bằng Python

Phân tích dữ liệu Apple Health bằng Python

Cách loại bỏ phần tử trùng lặp khỏi danh sách (List) trong Python

Cách loại bỏ phần tử trùng lặp khỏi danh sách (List) trong Python

Cách cắt (slicing) chuỗi trong Python

Cách cắt (slicing) chuỗi trong Python

Top