bitWalk's: QThreadPool と QRunnable (2) 〜 PySide6 〜

2024-02-23

QThreadPool と QRunnable (2) 〜 PySide6 〜

PySide (Qt for Python) は、Qt（キュート）の Python バインディングで、GUI などを構築するためのクロスプラットフォームなライブラリです。Linux/X11, macOS および Microsoft Windows をサポートしています。配布ライセンスは LGPL で公開されています。

以前、当ブログで PySide6 の QThreadPool と QRunnable を利用したマルチスレッディングのサンプルを紹介しました^[1]。これは、他サイトで紹介されていたサンプル [2] を自分用に判りやすく書き直したサンプルでした。

その後、意外とマルチスレッディングを必要とする機会が多く、最近の本ブログ記事でも QThreadPool をタイトルにした記事 [3] を掲載しました。しかし、あとで読み返してみるど、タイトルの QThreadPool を使うことよりも、どちらかというと mplfinance モジュールに興味が向くような内容になっていました。

今回は（主に自分自身が）テンプレート的に利用できるシンプルな QThreadPool と QRunnable のサンプルを、備忘録としてまとめました。

下記の OS 環境で動作確認をしています。

	Fedora Workstation 39	x86_64
	Python	3.12.1
	PySide6	6.6.2

サンプルを以下に示しました。

qt_threadpool_2.py

	#!/usr/bin/env python
	# coding: utf-8

	import sys
	import time

	from PySide6.QtCore import (
	QObject,
	QRunnable,
	QThreadPool,
	Signal,
	)
	from PySide6.QtWidgets import (
	QApplication,
	QMainWindow,
	QProgressBar,
	QPushButton,
	QStatusBar,
	)


	class WorkerSignal(QObject):
	threadFinished = Signal()
	threadProgress = Signal(int)


	class Worker(QRunnable, WorkerSignal):
	def __init__(self):
	super().__init__()

	def run(self):
	for i in range(100):
	time.sleep(0.1)
	self.threadProgress.emit(i + 1)
	self.threadFinished.emit()


	class Example(QMainWindow):
	def __init__(self):
	super().__init__()
	self.setWindowTitle('QThreadPool')
	self.threadpool = QThreadPool()

	self.btn = QPushButton('START')
	self.btn.setStyleSheet("""
	QPushButton:disabled {color: gray; background-color: lightgray;}
	""")
	self.btn.clicked.connect(self.button_clicked)
	self.setCentralWidget(self.btn)

	statusbar = QStatusBar()
	self.setStatusBar(statusbar)

	self.pbar = QProgressBar()
	self.pbar.setRange(0, 100)
	statusbar.addPermanentWidget(self.pbar, stretch=1)

	def button_clicked(self):
	self.btn.setEnabled(False)
	worker = Worker()
	worker.threadProgress.connect(self.on_status_update)
	worker.threadFinished.connect(self.on_completed)
	self.threadpool.start(worker)

	def on_completed(self):
	self.btn.setEnabled(True)
	self.pbar.reset()

	def on_status_update(self, progress: int):
	self.pbar.setValue(progress)


	def main():
	app = QApplication(sys.argv)
	ex = Example()
	ex.show()
	sys.exit(app.exec())


	if __name__ == '__main__':
	main()

view raw qt_threadpool_2.py hosted with ❤ by GitHub

このサンプルの実行例を以下に示しました。

START ボタンをクリックすると、Worker クラスのインスタンスの処理 worker.run() は、QThreadPool クラスのインスタンス self.threadpool の start メソッドで別スレッドで開始されます。

   :
   :
worker = Worker()
worker.threadProgress.connect(self.on_status_update)
worker.threadFinished.connect(self.on_completed)
self.threadpool.start(worker)

Worker クラスは QRunnable を継承したクラスで、run メソッドを実装します。QRunnable クラスでは Singal を利用できないので、別に QObject クラスを継承した WorkerSignal クラスで必要な Signal を定義しておき、これも Worker クラスが QRunnable と一緒に継承（多重継承）します。

class WorkerSignal(QObject):
    threadFinished = Signal()
    threadProgress = Signal(int)
 
 
class Worker(QRunnable, WorkerSignal):
    def __init__(self):
        super().__init__()
 
    def run(self):
        for i in range(100):
            time.sleep(0.1)
            self.threadProgress.emit(i + 1)
        self.threadFinished.emit()