8.10. Queue --- 同期キュークラス¶
ソースコード: Lib/Queue.py
Queue モジュールは、複数プロデューサ-複数コンシューマ (multi-producer, multi-consumer) キューを実装します。これは、複数のスレッドの間で情報を安全に交換しなければならないときのマルチスレッドプログラミングで特に有益です。このモジュールの Queue クラスは、必要なすべてのロックセマンティクスを実装しています。これはPythonのスレッドサポートの状況に依存します。 threading モジュールを参照してください。
このモジュールでは3種類のキューが実装されています。それらはキューから取り出されるエントリの順番だけが違います。 FIFOキューでは、最初に追加されたエントリが最初に取り出されます。 LIFOキューでは、最後に追加されたエントリが最初に取り出されます(スタックのように振る舞います)。 優先順位付きキュー(priority queue)では、エントリは(heapq モジュールを利用して)ソートされ、 最も低い値のエントリが最初に取り出されます。
Queue モジュールは以下のクラスと例外を定義します:
-
class
Queue.Queue(maxsize=0)¶ FIFOキューのコンストラクタです。maxsize はキューに置くことのできる要素数の上限を設定する整数です。いったんこの大きさに達したら、挿入はキューの要素が消費されるまでブロックされます。もし maxsize が0以下であるならば、キューの大きさは無限です。
-
class
Queue.LifoQueue(maxsize=0)¶ LIFOキューのコンストラクタです。maxsize はキューに置くことのできる要素数の上限を設定する整数です。いったんこの大きさに達したら、挿入はキューの要素が消費されるまでブロックされます。もし maxsize が0以下であるならば、キューの大きさは無限です。
バージョン 2.6 で追加.
-
class
Queue.PriorityQueue(maxsize=0)¶ 優先順位付きキューのコンストラクタです。maxsize はキューに置くことのできる要素数の上限を設定する整数です。いったんこの大きさに達したら、挿入はキューの要素が消費されるまでブロックされます。もし maxsize が0以下であるならば、キューの大きさは無限です。
最小の値を持つ要素が最初に検索されます (最小の値を持つ値は、
sorted(list(entries))[0]によって返されるものです)。典型的な要素のパターンは、(priority_number, data)形式のタプルです。バージョン 2.6 で追加.
-
exception
Queue.Empty¶ 空な
Queueオブジェクトで、非ブロックメソッドとしてget()(またはget_nowait()) が呼ばれたとき、送出される例外です。
-
exception
Queue.Full¶ 満杯な
Queueオブジェクトで、非ブロックメソッドとしてput()(またはput_nowait()) が呼ばれたとき、送出される例外です。
参考
collections.deque は、ロックなしで popleft() や append() といったアトミック操作が可能なキューの実装です。
8.10.1. キューオブジェクト¶
キューオブジェクト (Queue, LifoQueue, PriorityQueue) は、以下のpublicメソッドを提供しています。
-
Queue.qsize()¶ キューの近似サイズを返します。ここで、qsize() > 0 は後続の get() がブロックしないことを保証しないこと、また qsize() < maxsize が put() がブロックしないことを保証しないことに注意してください。
-
Queue.empty()¶ キューが空の場合は
Trueを返し、そうでなければFalseを返します。empty() がTrueを返しても、後続の put() の呼び出しがブロックしないことは保証されません。同様に、empty() がFalseを返しても、後続の get() の呼び出しがブロックしないことは保証されません。
-
Queue.full()¶ キューが一杯の場合は
Trueを返し、そうでなければFalseを返します。full() がTrueを返しても、後続の get() の呼び出しがブロックしないことは保証されません。同様に、full() がFalseを返しても、後続の put() の呼び出しがブロックしないことは保証されません。
-
Queue.put(item[, block[, timeout]])¶ item をキューに入れます。 もしオプション引数 block が真で timeout が
None(デフォルト) の場合は、必要であればフリースロットが利用可能になるまでブロックします。 timeout が正の数の場合は、最大で timeout 秒間ブロックし、その時間内に空きスロットが利用可能にならなければ、例外Fullを送出します。 そうでない場合 (block が偽) は、空きスロットが直ちに利用できるならば、キューにアイテムを置きます。 できないならば、例外Fullを送出します (この場合 timeout は無視されます)。バージョン 2.3 で追加: timeout パラメータが追加されました。
-
Queue.put_nowait(item)¶ put(item, False)と等価です。
-
Queue.get([block[, timeout]])¶ キューからアイテムを取り除き、それを返します。 オプション引数 block が真で timeout が
None(デフォルト) の場合は、必要であればアイテムが取り出せるようになるまでブロックします。 もし timeout が正の数の場合は、最大で timeout 秒間ブロックし、その時間内でアイテムが取り出せるようにならなければ、例外Emptyを送出します。 そうでない場合 (block が偽) は、直ちにアイテムが取り出せるならば、それを返します。 できないならば、例外Emptyを送出します (この場合 timeout は無視されます)。バージョン 2.3 で追加: timeout パラメータが追加されました。
-
Queue.get_nowait()¶ get(False)と等価です。
キューに入れられたタスクが全てコンシューマスレッドに処理されたかどうかを追跡するために 2つのメソッドが提供されます。
-
Queue.task_done()¶ 過去にキューに入れられたタスクが完了した事を示します。キューのコンシューマスレッドに利用されます。タスクの取り出しに使われた、各
get()に対して、それに続くtask_done()の呼び出しは、取り出したタスクに対する処理が完了した事をキューに教えます。join()がブロックされていた場合、全itemが処理された (キューにput()された全てのitemに対してtask_done()が呼び出されたことを意味します) 時に復帰します。キューにある要素より多く呼び出された場合
ValueErrorが発生します。バージョン 2.5 で追加.
-
Queue.join()¶ キューの中の全アイテムが処理される間でブロックします。
キューにitemが追加される度に、未完了タスクカウントが増やされます。コンシューマスレッドが
task_done()を呼び出して、itemを受け取ってそれに対する処理が完了した事を知らせる度に、未完了タスクカウントが減らされます。未完了タスクカウントが0になったときに、join()のブロックが解除されます。バージョン 2.5 で追加.
キューに入れたタスクが完了するのを待つ例:
def worker():
while True:
item = q.get()
do_work(item)
q.task_done()
q = Queue()
for i in range(num_worker_threads):
t = Thread(target=worker)
t.daemon = True
t.start()
for item in source():
q.put(item)
q.join() # block until all tasks are done