Pythonのqueueモジュールで学ぶスタックとキューの基本と使い方

Pythonでは、スタックやキューといったデータ構造を非常に簡単に実装できます。スタックは「後入れ先出し(LIFO: Last-In, First-Out)」の原理で動作することからLIFOと呼ばれ、キューは「先入れ先出し(FIFO: First-In, First-Out)」の原理で動作することからFIFOと呼ばれます。Pythonに組み込まれたモジュールや関数を活用すれば、コードを短くシンプルに保つことができます。

queueモジュールは、マルチプロデューサ・マルチコンシューマ型のキューを実装したものであり、複数のスレッド間で情報を安全にやり取りする必要があるスレッドプログラミングにおいて特に有用です。このモジュールのQueueクラスは、必要なロック処理(ロッキングセマンティクス)をすべて内部で実装しており、Pythonのスレッドサポートに依存して動作します。

このモジュールでは、要素を取り出す順序だけが異なる3種類のキューが提供されています。

• FIFOキュー:最初に追加されたタスクが最初に取り出されます。
• LIFOキュー:最も新しく追加された要素が最初に取り出されます(スタックと同じ動作)。
• 優先度付きキュー(PriorityQueue):要素は常にソートされた状態で保持され(heapqモジュールを使用)、最も値の小さい要素から順に取り出されます。

以下、queueモジュールで定義されている主なクラスと例外について詳しく見ていきましょう。

Queue.Queue(maxsize=0)

FIFOキューのコンストラクタです。引数maxsizeには、キューに格納できる項目数の上限を指定する整数を渡します。このサイズに達すると、キュー内の項目が消費されるまで挿入操作はブロックされます。maxsizeが0以下の場合、キューのサイズは無限になります。

Queue.LifoQueue(maxsize=0)

LIFOキュー(スタック)のコンストラクタです。引数maxsizeの意味と挙動はFIFOキューと同様です。

Queue.PriorityQueue(maxsize=0)

優先度付きキューのコンストラクタです。引数maxsizeの意味と挙動は他のキューと同じです。

例外 Queue.Empty

空のQueueオブジェクトに対して、ブロックなしのget()(またはget_nowait())を呼び出した際に発生する例外です。

例外 Queue.Full

満杯のQueueオブジェクトに対して、ブロックなしのput()(またはput_nowait())を呼び出した際に発生する例外です。

Queueオブジェクトの主なメソッド

qsize()

キューのおおよそのサイズ(現在格納されている項目数)を返します。

empty()

キューが空であればTrueを、そうでなければFalseを返します。なお、empty()がTrueを返しても、その後のput()がブロックしないことは保証されません。同様に、Falseを返してもget()がブロックしない保証はありません。

full()

キューが満杯であればTrueを、そうでなければFalseを返します。full()がTrueを返しても、その後のget()がブロックしない保証はなく、Falseを返してもput()がブロックしない保証はありません。

put(item[, block[, timeout]])

キューにitemを追加します。オプション引数blockがTrueでtimeoutがNone(デフォルト)の場合、空きスロットが利用可能になるまで必要に応じてブロックします。timeoutに正の数を指定した場合は最大でその秒数だけブロックし、時間内に空きスロットが得られなければFull例外を送出します。blockがFalseの場合、即座に空きスロットがあれば項目を追加し、なければFull例外を送出します(この場合timeoutは無視されます)。

get([block[, timeout]])

キューから項目を取り除いて返します。オプション引数blockがTrueでtimeoutがNone(デフォルト)の場合、項目が利用可能になるまで必要に応じてブロックします。timeoutに正の数を指定した場合は最大でその秒数だけブロックし、時間内に項目が得られなければEmpty例外を送出します。blockがFalseの場合、即座に項目があればそれを返し、なければEmpty例外を送出します(この場合timeoutは無視されます)。

task_done()

以前にキューへ追加されたタスクの処理が完了したことを示します。キューのコンシューマ(取り出し側)スレッドで使用されます。get()でタスクを取得するたびに、対応するtask_done()を呼び出すことで、そのタスクの処理が完了したことをキューに伝えます。

join()が現在ブロック中の場合、すべての項目の処理が完了すると(つまりput()された全項目に対してtask_done()が呼び出されると)ブロックが解除されます。

キューに追加された項目数より多くtask_done()を呼び出すと、ValueErrorが発生します。

join()

キュー内のすべての項目が取得され、処理されるまでブロックします。

未完了タスクのカウントは、項目がキューに追加されるたびに1ずつ増加します。そして、コンシューマスレッドがtask_done()を呼び出して項目の取得と処理の完了を示すたびに減少していきます。未完了タスクのカウントがゼロになると、join()のブロックが解除されます。

サンプルコード:FIFOキューの基本

import queue
# キューの最大容量を40に設定
Q = queue.Queue(maxsize=40)
Q.put(50)
Q.put(90)
Q.put(10)
Q.put(70)
print(Q.get())
print(Q.get())
print(Q.get())
print(Q.get())

実行結果

50
90
10
70

FIFOキューでは、投入した順番どおりにデータが取り出されていることがわかります。

サンプルコード:アンダーフローとオーバーフローの確認

import queue
Q = queue.Queue(maxsize=30)
print(Q.qsize())
Q.put(50)
Q.put(90)
Q.put(10)
Q.put(70)
print("Full: ", Q.full())
Q.put(90)
Q.put(100)
print("Full: ", Q.full())
print(Q.get())
print(Q.get())
print(Q.get())
print("Empty: ", Q.empty())
print(Q.get())
print(Q.get())
print(Q.get())
print("Empty: ", Q.empty())
print("Full: ", Q.full())

実行結果

0
Full: False
Full: False
50
90
10
Empty: False
70
90
100
Empty: True
Full: False

この例では、full()やempty()を使ってキューの状態を確認しています。すべての要素を取り出し終えるとempty()がTrueを返すことが確認できます。

サンプルコード:LifoQueueによるスタックの実装

import queue
S = queue.LifoQueue(maxsize=10)
# qsize()で現在のサイズを取得
print(S.qsize())
S.put(50)
S.put(90)
S.put(10)
S.put(70)
S.put(90)
S.put(10)
print("Full: ", S.full())
print("Size: ", S.qsize())
# データはキューとは逆の順序で取り出される
print(S.get())
print(S.get())
print(S.get())
print(S.get())
print(S.get())
print("Empty: ", S.empty())

実行結果

0
Full: False
Size: 6
10
90
70
10
90
Empty: False

LifoQueueでは、最後に追加したデータから逆順に取り出されます。これがスタックとしての典型的な動作です。マルチスレッド環境でのタスク管理や、履歴の巻き戻しなど、さまざまな場面で活用できるので、ぜひ実際にコードを書いて挙動を確かめてみてください。