Pythonでマルチスレッドキューを実装する方法を徹底解説

はじめに

本記事では、実行すべきタスクをすべて保持する「タスクキュー」と、そのキューと連携して各要素を個別に処理する「スレッドプール」を作成する方法を解説します。

まず、「キューとは何か?」という基本的な疑問から始めましょう。キューとは、異なる要素の集合体を特定の順序で管理するデータ構造のことです。身近な例を挙げて説明します。

スーパーのレジで会計のために列に並んでいる場面を想像してください(どこのスーパーかは気にしないでください)。

会計待ちの人々の列には、次のような特徴があります。

• 人は列の一方の端から入り、もう一方の端から出ていく。
• 人Aが人Bより先に列に入ったなら、人Aは人Bより先に列から出る(ただし、人Bが有名人であるか、優先権を持っている場合は別)。
• 全員が会計を済ませれば、列に残っている人はいなくなる。

さて、プログラミングの世界に戻ると、キューもまったく同じように動作します。

• enqueue(エンキュー) — 要素をキューの末尾に追加する。
• dequeue(デキュー) — 要素をキューの先頭から取り除く。

さらに、次の2つの方式があります。FIFO(First In First Out:先入れ先出し)は、最初に追加された要素が最初に取り除かれる方式です。一方LIFO(Last In First Out:後入れ先出し)は、最後に追加された要素が最初に取り除かれる方式です。

Pythonにおけるキューデータ構造の実装方法

Pythonのqueueモジュールは、キューデータ構造のシンプルな実装を提供しています。各キューでは、以下のメソッドを使用できます。

• get():次の要素を返します。
• put():新しい要素を追加します。
• qsize():キュー内の現在の要素数を返します。
• empty():キューが空かどうかを示す真偽値を返します。
• full():キューが満杯かどうかを示す真偽値を返します。

ステップ1:乗算を行う関数を作成する

まず、引数xを受け取り、1からxまでの数値を順番に掛け合わせる関数を作成します。例えば、この関数に5を渡すと、1〜5まで順に5を掛けていき(1×5、2×5、3×5、4×5、5×5)、結果をリストとして返します。

def print_multiply(x):
    output_value = []
    for i in range(1, x + 1):
        output_value.append(i * x)
    print(f"Output \n *** {x} の乗算結果は - {output_value}")

print_multiply(5)

実行結果

*** 5 の乗算結果は - [5, 10, 15, 20, 25]

ステップ2:キューを処理する関数を作成する

次に、process_queue()という関数を作成します。この関数は、キューオブジェクトから次の要素を取得しようと試みます。ロジックは非常にシンプルで、キューが空になるまで要素を処理し続けるというものです。処理の進行を少し遅らせるためにsleepを使用します。

def process_queue():
    while True:
        try:
            value = my_queue.get(block=False)
        except queue.Empty:
            return
        else:
            print_multiply(value)
            time.sleep(2)

ステップ3:スレッドクラスを作成する

続いて、クラスを作成します。新しいインスタンスが初期化されて開始されると、process_queue()関数が呼び出される仕組みです。

class MultiThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name

    def run(self):
        print(f" ** スレッド開始 - {self.name}")
        process_queue()
        print(f" ** スレッド完了 - {self.name}")

ステップ4:入力リストをキューに格納する

最後に、数値の入力リストを受け取ってキューに格納します。

# 変数の設定
input_values = [5, 10, 15, 20]

# キューへの格納
my_queue = queue.Queue()
for x in input_values:
    my_queue.put(x)

ステップ5:すべてを組み合わせる

それでは、これまでのコードをすべて統合しましょう。

import queue
import threading
import time

# スレッドクラス
class MultiThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name

    def run(self):
        print(f"Output \n ** スレッド開始 - {self.name}")
        process_queue()
        print(f" ** スレッド完了 - {self.name}")

# キューの処理
def process_queue():
    while True:
        try:
            value = my_queue.get(block=False)
        except queue.Empty:
            return
        else:
            print_multiply(value)
            time.sleep(2)

# 乗算関数
def print_multiply(x):
    output_value = []
    for i in range(1, x + 1):
        output_value.append(i * x)
    print(f" \n *** {x} の乗算結果は - {output_value}")

# 入力変数
input_values = [2, 4, 6, 5, 10, 3]

# キューへの格納
my_queue = queue.Queue()
for x in input_values:
    my_queue.put(x)

# 4つのスレッドを初期化して開始
thread1 = MultiThread('First')
thread2 = MultiThread('Second')
thread3 = MultiThread('Third')
thread4 = MultiThread('Fourth')

# スレッドの開始
thread1.start()
thread2.start()
thread3.start()
thread4.start()

# スレッドの終了待機
thread1.join()
thread2.join()
thread3.join()
thread4.join()

実行結果

** スレッド開始 - First
*** 2 の乗算結果は - [2, 4]

** スレッド開始 - Second
*** 4 の乗算結果は - [4, 8, 12, 16]

** スレッド開始 - Third
*** 6 の乗算結果は - [6, 12, 18, 24, 30, 36]

** スレッド開始 - Fourth
*** 5 の乗算結果は - [5, 10, 15, 20, 25]
*** 10 の乗算結果は - [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
*** 3 の乗算結果は - [3, 6, 9] ** スレッド完了 - Third
** スレッド完了 - Fourth
** スレッド完了 - Second ** スレッド完了 - First

まとめ

以上で、キューの概念を実装することができました。注目すべき点として、スレッドは4つしかありませんが、処理すべき値は6つあります。そのため、先にキューに到着したタスクから順に実行され、残りのタスクは他の処理が完了するのを待ちながら列に並ぶことになります。

これは現実の生活にも似ています。例えば、レジが3つあるのに会計待ちの人が10人いる場合、10人の客は3つの列に分かれて並び、会計を終えた人から順に列を離れ、次の人に道を譲っていくのです。