PyQtQThreadPool

Resumo : neste tutorial, você aprenderá como criar um aplicativo multithreading PyQt que usa QThreadPoolclasses QRunnable.

Introdução às classes QThreadPool e QRunnable

A QThreadclasse permite transferir uma tarefa de longa execução para um thread de trabalho para tornar o aplicativo mais responsivo. A classe QThread funciona bem se o aplicativo tiver alguns threads de trabalho.

Um programa multithread é eficiente quando possui um número de QThreadobjetos que corresponde ao número de núcleos da CPU.

Além disso, a criação de threads é bastante cara em termos de recursos do computador. Portanto, o programa deve reutilizar os threads criados tanto quanto possível.

Portanto, usar a QThreadclasse para gerenciar threads de trabalho apresenta dois desafios principais:

  • Determine o número ideal de threads para o aplicativo com base no número de núcleos da CPU.
  • Reutilize e recicle os threads tanto quanto possível.

Felizmente, o PyQt tem QThreadPoolaulas que resolvem esses desafios para você. A QThreadPoolclasse é frequentemente usada com a QRunnableclasse.

  • A QRunnableclasse representa uma tarefa que você deseja executar em um thread de trabalho.
  • O QThreadPoolexecuta um QRunnableobjeto e gerencia e recicla threads automaticamente.

Cada aplicação Qt possui um QThreadPoolobjeto global que pode ser acessado através do globalInstance()método estático da QThreadPoolclasse.

Para usar as classes QThreadPoole QRunnable, siga estas etapas:

Primeiro, crie uma classe que herde da QRunnableclasse e substitua o run()método:

class Worker(QRunnable):
    @Slot()
    def run(self):
        # place a long-running task here
        passLinguagem de código:  Python  ( python )

Segundo, acesse o pool de threads na janela principal e inicie os threads de trabalho:

class MainWindow(QMainWindow):
    # other methods
    # ...

    def start(self):
        """ Create and execute worker threads
        """
        pool = QThreadPool.globalInstance()
        for _ in range(1, 100):
            pool.start(Worker())Linguagem de código:  Python  ( python )

Para atualizar o progresso do trabalhador para o thread principal, você usa sinais e slots. No entanto, o QRunnablenão suporta sinal.

Portanto, você precisa definir uma classe separada que herde QObjecte use essa classe na classe Worker. Aqui estão as etapas:

Primeiro, defina a Signalsclasse que subclassifica a QObjectclasse:

class Signals(QObject):
    completed = Signal()Linguagem de código:  Python  ( python )

Na Signalsclasse, definimos um sinal chamado completed. Observe que você pode definir quantos sinais forem necessários.

Segundo, emita o completedsinal quando o trabalho for concluído na Workeraula:

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

    @Slot()
    def run(self):
        # long running task
        # ...
        # emit the completed signal
        self.signals.completed.emit()Linguagem de código:  Python  ( python )

Terceiro, conecte o sinal do thread de trabalho a um slot da janela principal antes de enviar o trabalhador ao pool:

class MainWindow(QMainWindow):
    # other methods
    # ...

    def start(self):
        """ Create and execute worker threads
        """
        pool = QThreadPool.globalInstance()
        for _ in range(1, 100):
            worker = Worker()
            worker.signals.completed.connect(self.update)
            pool.start(worker)

    def update(self):
        # update the worker
        passLinguagem de código:  Python  ( python )

Exemplo de QThreadPool

O exemplo a seguir ilustra como usar a classe QThreadPooland 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())


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)Linguagem de código:  Python  ( python )

Saída:

Classe de sinais

Defina a classe Signals que herda da QObjectclasse para suportar sinais. Na Signalsaula, definimos dois sinais:

  • O startedsinal será emitido quando um trabalhador for iniciado.
  • O completedsinal será emitido quando um trabalhador for concluído.

Ambos os sinais aceitam um número inteiro que identifica o número do trabalho:

class Signals(QObject):
    started = Signal(int)
    completed = Signal(int)Linguagem de código:  Python  ( python )

Classe trabalhadora

A Workerclasse herda da QRunnableclasse. A Workerclasse representa uma tarefa de longa execução que transferimos para um thread de trabalho:

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)Linguagem de código:  Python  ( python )

Primeiro, inicialize o número do trabalho (n) e o objeto Signals no __init__()método.

Segundo, substitua o run()método da QRunnableclasse. Para simular uma tarefa de longa duração, usamos a sleep()função do módulo de tempo. Antes de iniciar o cronômetro, emitimos o sinal de início; após a conclusão do cronômetro, emitimos o sinal concluído.

Classe MainWindow

A MainWindowclasse define o UIpara o aplicativo:

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

        self.setWindowTitle('QThreadPool Demo')

        self.comleted_jobs = []
        self.job_count = 10

        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):
            runnable = Worker(i)
            runnable.signals.completed.connect(self.complete)
            runnable.signals.started.connect(self.start)
            pool.start(runnable)

    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)Linguagem de código:  Python  ( python )

Primeiro, inicialize o número de jobs ( job_count) e completed_jobsliste no __init__()método da MainWindowclasse:

self.job_count = 10
self.comleted_jobs = []Linguagem de código:  Python  ( python )

Segundo, defina o start_jobs()método que será executado quando o usuário clicar no botão iniciar:

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)Linguagem de código:  Python  ( python )

O restart()redefine o completed_jobs, atualiza a barra de progresso para zero e desativa o botão Iniciar:

def restart(self):
    self.progress_bar.setValue(0)
    self.comleted_jobs = []
    self.btn_start.setEnabled(False)Linguagem de código:  Python  ( python )

Para obter o QThreadPoolobjeto, usamos o globalInstance()da QThreadPoolclasse:

pool = QThreadPool.globalInstance()Linguagem de código:  Python  ( python )

Criamos vários trabalhadores, conectamos seus sinais aos métodos da MainWindowclasse e iniciamos threads de trabalho usando o start()método do QThreadPoolobjeto.

O start()método adiciona a mensagem que inicia um thread de trabalho ao QListWidget:

def start(self, n):
    self.list.addItem(f'Job #{n} started...')Linguagem de código:  Python  ( python )

O completed()método é executado sempre que um thread de trabalho é concluído. Ele adiciona uma mensagem ao QListWidget, atualiza a barra de progresso e ativa o botão Iniciar se todos os threads de trabalho forem concluídos:

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)Linguagem de código:  Python  ( python )

Usando QThreadPool para obter preços de ações

O seguinte programa de listagem de ações lê símbolos de ações do symbols.txtarquivo e usa QThreadPoolpara obter os preços das ações do site Yahoo Finance:

Programa de listagem de ações:

import sys
from pathlib import Path

from PyQt6.QtCore import QRunnable, Qt, QObject, QThreadPool, pyqtSignal as Signal, pyqtSlot as Slot
from PyQt6.QtWidgets import QApplication,  QMainWindow, QPushButton, QWidget, QGridLayout, QProgressBar, QTableWidget, QTableWidgetItem
from PyQt6.QtGui import QIcon

from lxml import html
import requests


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


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})


class Window(QMainWindow):
    def __init__(self, filename, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.symbols = self.read_symbols(filename)

        self.results = []

        self.setWindowTitle('Stock Listing')
        self.setGeometry(100, 100, 400, 300)
        self.setWindowIcon(QIcon('./assets/stock.png'))

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

        # set up button & progress bar
        self.btn_start = QPushButton('Get Prices', clicked=self.get_prices)
        self.progress_bar = QProgressBar(minimum=1, maximum=len(self.symbols))

        # set up table widget
        self.table = QTableWidget(widget)
        self.table.setColumnCount(2)
        self.table.setColumnWidth(0, 150)
        self.table.setColumnWidth(1, 150)

        self.table.setHorizontalHeaderLabels(['Symbol', 'Price'])

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

        # show the window
        self.show()

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

    def reset_ui(self):
        self.progress_bar.setValue(1)
        self.table.setRowCount(0)

    def get_prices(self):
        # reset ui
        self.reset_ui()

        # start worker threads
        pool = QThreadPool.globalInstance()
        stocks = [Stock(symbol) for symbol in self.symbols]
        for stock in stocks:
            stock.signal.completed.connect(self.update)
            pool.start(stock)

    def update(self, data):
        # add a row to the table
        row = self.table.rowCount()
        self.table.insertRow(row)
        self.table.setItem(row, 0, QTableWidgetItem(data['symbol']))
        self.table.setItem(row, 1, QTableWidgetItem(str(data['price'])))

        # update the progress bar
        self.progress_bar.setValue(row + 1)

        # sort the list by symbols once completed
        if row == len(self.symbols) - 1:
            self.table.sortItems(0, Qt.SortOrder.AscendingOrder)


if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = Window('symbols.txt')
    sys.exit(app.exec())Linguagem de código:  Python  ( python )

Como funciona.

Classe de sinais

Definimos a Signalsclasse que é uma subclasse do QObject. A classe Signals possui uma variável de classe concluída que é uma instância da Signalclasse.

O sinal concluído contém um dicionário e é emitido assim que o programa conclui a obtenção do preço das ações.

class Signals(QObject):
    completed = Signal(dict)Linguagem de código:  Python  ( python )

Classe de ações

A STockclasse herda da QRunnableclasse. Ele substitui o run()método que obtém o preço das ações do site do Yahoo Finance.

Depois de concluído, o run()método emite o sinal concluído com o símbolo da ação e o preço.

Se ocorrer um erro como o símbolo não ser encontrado ou o site alterar a forma como exibe o preço da ação, o run()método retornará o símbolo com o preço como uma string 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})Linguagem de código:  Python  ( python )

Observe que o Yahoo Finance pode alterar sua estrutura. Para que o programa funcione, você precisa alterar o XPath do preço do novo:

//*[@id="quote-header-info"]/div[3]/div[1]/div[1]/fin-streamer[1]/text()Linguagem de código:  Python  ( python )

Classe MainWindow

Primeiro, leia os símbolos de um arquivo e atribua-os às self.symbolsvariáveis:

 self.symbols = self.read_symbols(filename)Linguagem de código:  Python  ( python )

O read_symbols()método é assim:

def read_symbols(self, filename):
    path = Path(filename)
    text = path.read_text()
    return [symbol.strip() for symbol in text.split('\n')]Linguagem de código:  Python  ( python )

O arquivo de texto ( symbols.txt) contém cada símbolo por linha:

AAPL	
MSFT
GOOG	
AMZN
TSLA
META
NVDA
BABA
CRM
INTC
PYPL	
AMD
ATVI
EA
TTD
ORCLLinguagem de código:  Python  ( python )

Segundo, defina o get_pricesque usa QThreadPoolpara criar threads de trabalho para obter preços de ações:

def get_prices(self):
    # reset ui
    self.reset_ui()

    # start worker threads
    pool = QThreadPool.globalInstance()
    stocks = [Stock(symbol) for symbol in self.symbols]
    for stock in stocks:
        stock.signal.completed.connect(self.update)
        pool.start(stock)Linguagem de código:  Python  ( python )

O reset_ui()método limpa todas as linhas QTableWidgete define a barra de progresso para seu valor mínimo:

def reset_ui(self):
    self.table.setRowCount(0)
    self.progress_bar.setValue(1)Linguagem de código:  Python  ( python )

Terceiro, defina o update()método que será chamado assim que cada thread de trabalho for concluído. O update()método adiciona uma nova linha à tabela, atualiza a barra de progresso e classifica os símbolos quando todos os threads de trabalho são concluídos:

def update(self, data):
    # add a row to the table
    row = self.table.rowCount()
    self.table.insertRow(row)
    self.table.setItem(row, 0, QTableWidgetItem(data['symbol']))
    self.table.setItem(row, 1, QTableWidgetItem(str(data['price'])))

    # update the progress bar
    self.progress_bar.setValue(row + 1)

    # sort the list by symbols once completed
    if row == len(self.symbols) - 1:
        self.table.sortItems(0, Qt.SortOrder.AscendingOrder)Linguagem de código:  Python  ( python )

Resumo

  • Use a QRunnableclasse para representar uma tarefa de longa execução que será transferida para um thread de trabalho.
  • Use a QThreadPoolclasse para gerenciar threads de trabalho automaticamente.
  • Cada aplicativo PyQt possui um QThreadPoolobjeto. Use o globalInstance()método para obter o QThreadPoolobjeto global.
  • Use o start()método do QThreadPoolobjeto para iniciar um thread de trabalho.
PyQtQThread
Anterior
PyQtQThread

Próximo
Folhas de estilo Qt

Folhas de estilo Qt

Deixe um comentário

O seu endereço de email não será publicado. Campos obrigatórios marcados com *