Pythonで優先度付きキュー（PriorityQueue）を実装する方法を徹底解説

はじめに

Pythonのqueueモジュールは、マルチスレッドプログラミングに適したFIFO（First-In, First-Out：先入れ先出し）およびLIFO（Last-In, First-Out：後入れ先出し）のデータ構造を提供します。キューを使うことで、セッション情報やパス、変数など、さまざまなデータをスレッド間で安全に受け渡すことができます。ロック処理は通常モジュール側が自動的に行ってくれるため、開発者はスレッドセーフな実装を意識せずに済みます。

注意： 本記事では、キューの基本的な仕組みをすでに理解していることを前提としています。もし不安であれば、先に公式ドキュメントなどの参考資料に目を通しておくことをおすすめします。

1. 基本的なFIFOキューの実装

まずは最もシンプルなFIFOキューから見ていきましょう。要素は追加された順番通りに取り出されます。

import queue
fifo = queue.Queue()

# キューに数値を追加
for i in range(5):
    fifo.put(i)

# キューが空でなければ数値を取り出す
print(f"Output \n")
while not fifo.empty():
    print(f" {fifo.get()} ")

実行結果

0
1
2
3
4

この例では単一スレッドを使用しており、挿入した順序と同じ順序で要素が取り出されることが確認できます。これがFIFOの基本動作です。

2. 基本的なLIFOキューの実装

次にLIFOキューを実装してみましょう。queue.LifoQueue()を使うことで、スタックのように「最後に追加した要素」から順に取り出されます。

import queue
lifo = queue.LifoQueue()

# キューに数値を追加
for i in range(5):
    lifo.put(i)

print(f"Output \n")
# キューが空でなければ数値を取り出す
while not lifo.empty():
    print(f" {lifo.get()} ")

実行結果

4
3
2
1
0

この例からわかるように、get()メソッドは最後にキューへ投入された要素を最初に返します。これがLIFO（スタック）の動作です。

3. 優先度付きキュー（PriorityQueue）の実装

最後に、本題である優先度付きキューの実装方法を見ていきます。

場合によっては、キュー内の項目を作成・追加した順序ではなく、その優先度に基づいて処理順序を決めたいことがあります。たとえば、本番環境で稼働しているビジネス上重要なジョブは、開発者が印刷したいだけのジョブよりも高いCPU優先度と処理順位を持つべきです。PriorityQueueは、キューの中身のソート順序に従って、どの項目を取り出すかを決定します。

以下の例では、優先度と説明文を受け取るJobクラスを定義し、比較演算のための特殊メソッド（__eq__と__lt__）を実装しています。

import queue
import threading

# 優先度と説明を受け取り、優先度を検証するクラス
class Job:
    def __init__(self, priority, description):
        self.priority = priority
        self.description = description
        print('New job:', description)
        return

    def __eq__(self, other):
        try:
            return self.priority == other.priority
        except AttributeError:
            return NotImplemented

    def __lt__(self, other):
        try:
            return self.priority < other.priority
        except AttributeError:
            return NotImplemented

# 優先度付きキューを作成し、優先度を設定
q = queue.PriorityQueue()
q.put(Job(90, 'Developer-Print job'))
q.put(Job(2, 'Business-Report job'))
q.put(Job(1, 'Business-Critical Job'))

# ジョブを処理する関数
def process_job(q):
    while True:
        next_job = q.get()
        print(f" *** Now, Processing the job - {next_job.description}")
        q.task_done()

# ワーカースレッドを定義
workers = [
    threading.Thread(target=process_job, args=(q,)),
    threading.Thread(target=process_job, args=(q,))]

# スレッドを開始して終了を待機
for w in workers:
    w.setDaemon(True)
    w.start()

q.join()

実行結果（ジョブ登録時の出力）

New job: Developer-Print job
New job: Business-Report job
New job: Business-Critical Job

実行結果（ジョブ処理時の出力）

*** Now, Processing the job - Business-Critical Job
*** Now, Processing the job - Business-Report job
*** Now, Processing the job - Developer-Print job

まとめ

この例では、複数のワーカースレッドがジョブを消費しており、各ジョブはget()が呼び出された時点でのキュー内の優先度に基づいて処理されています。注目すべき点は、ジョブが追加された順序に関係なく、ビジネス上の重要度（優先度の数値が小さいほど高優先度）に従って処理順序が決まるという点です。

このようにqueue.PriorityQueueを活用すれば、マルチスレッド環境でも安全かつ柔軟にタスクの優先順位制御を行うことができます。バッチ処理やジョブスケジューラなど、優先度ベースのタスク管理が必要な場面でぜひ活用してみてください。