Pythonのマルチスレッド処理におけるキューと優先度付きキューの使い方

Queueモジュールとは

Pythonのqueueモジュールを使うと、複数のスレッド間で安全にデータをやり取りできるキューを作成できます。キューは最大で指定した数のアイテムを保持でき、スレッドセーフな設計になっているため、マルチスレッドプログラミングでもデータ競合を気にせず利用できます。

キューを操作するための主なメソッドは以下の通りです。

• get() − キューからアイテムを取り出して返します。
• put() − キューにアイテムを追加します。
• qsize() − 現在キューに入っているアイテムの数を返します。
• empty() − キューが空の場合はTrue、そうでなければFalseを返します。
• full() − キューが満杯の場合はTrue、そうでなければFalseを返します。

サンプルコード

以下は、3つのワーカースレッドが1つのキューからデータを取り出して処理する例です（Python 3系の記法）。

#!/usr/bin/python3
import queue
import threading
import time

exitFlag = 0

class myThread(threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q

    def run(self):
        print("Starting " + self.name)
        process_data(self.name, self.q)
        print("Exiting " + self.name)

def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print("%s processing %s" % (threadName, data))
        else:
            queueLock.release()
        time.sleep(1)

threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = queue.Queue(10)
threads = []
threadID = 1

# 新しいスレッドを作成
for tName in threadList:
    thread = myThread(threadID, tName, workQueue)
    thread.start()
    threads.append(thread)
    threadID += 1

# キューにデータを投入
queueLock.acquire()
for word in nameList:
    workQueue.put(word)
queueLock.release()

# キューが空になるまで待機
while not workQueue.empty():
    pass

# スレッドに終了を通知
exitFlag = 1

# すべてのスレッドの完了を待機
for t in threads:
    t.join()

print("Exiting Main Thread")

実行結果

上記のコードを実行すると、次のような結果が出力されます。

Starting Thread-1
Starting Thread-2
Starting Thread-3
Thread-1 processing One
Thread-2 processing Two
Thread-3 processing Three
Thread-1 processing Four
Thread-2 processing Five
Exiting Thread-3
Exiting Thread-1
Exiting Thread-2
Exiting Main Thread

コードの解説

このプログラムの処理の流れは以下の通りです。

1. スレッドの作成： myThreadクラスをもとに3つのワーカースレッドを生成し、開始します。
2. キューへの投入： メインスレッドがロックを取得し、「One」〜「Five」の5つのデータをキューに追加します。
3. データ処理： 各ワーカースレッドはロックを取得しながらキューからデータを取り出し、処理を行います。
4. 終了処理： キューが空になったらexitFlagを立ててスレッドに終了を通知し、join()ですべてのスレッドの完了を待ちます。

優先度付きキュー（PriorityQueue）について

優先キューを実現したい場合は、queue.PriorityQueueを使用します。通常のQueueがFIFO（先入れ先出し）方式であるのに対し、PriorityQueueは優先度の高いアイテムから順に取り出されます。優先度はタプルの最初の要素で指定するのが一般的です。

import queue

pq = queue.PriorityQueue()
pq.put((2, "低優先度のタスク"))
pq.put((1, "高優先度のタスク"))
pq.put((3, "最低優先度のタスク"))

while not pq.empty():
    print(pq.get()[1])

# 出力順: 高優先度のタスク → 低優先度のタスク → 最低優先度のタスク

このように、queueモジュールを活用することで、マルチスレッド環境でも安全かつ柔軟にタスク管理を行うことができます。