queue --- 同期キュークラス

ソースコード: Lib/queue.py

queue モジュールは、複数の生産者と消費者が使えるキューを実装しています。 複数のスレッド間で安全に情報を交換する必要があるスレッドプログラミングで特に役立ちます。 このモジュールの Queue クラスは、必要なロック処理をすべて実装しています。

このモジュールでは3種類のキューが実装されています。それらはキューから取り出されるエントリの順番だけが違います。 FIFO キューでは、最初に追加されたエントリが最初に取り出されます。 LIFO キューでは、最後に追加されたエントリが最初に取り出されます(スタックのように振る舞います)。 優先順位付きキュー(priority queue)では、エントリは(heapq モジュールを利用して)ソートされ、 最も低い値のエントリが最初に取り出されます。

内部的には、これらの3種類のキューは競争スレッドを一時的にブロックするためにロックを使っています; しかし、スレッド内での再入を扱うようには設計されていません。

さらに、このモジュールは「単純な」:abbr:FIFO (first-in, first-out) キュー型 SimpleQueue を実装しています。 この型は機能を絞る代わりに、その実装によって追加の保証を提供します。

queue モジュールは次のクラスと例外を定義しています。

class queue.Queue(maxsize=0)

FIFO キューのコンストラクタです。 maxsize はキューに入れられる要素数の上限を設定する整数です。 いったんこの大きさに達したら、挿入処理はキューの要素が消費されるまでブロックされます。 maxsize が0以下の場合は、キューの大きさは無限です。

class queue.LifoQueue(maxsize=0)

LIFO キューのコンストラクタです。 maxsize はキューに入れられる要素数の上限を設定する整数です。 いったんこの大きさに達したら、挿入処理はキューの要素が消費されるまでブロックされます。 maxsize が0以下の場合は、キューの大きさは無限です。

class queue.PriorityQueue(maxsize=0)

優先順位付きキューのコンストラクタです。maxsize はキューに置くことのできる要素数の上限を設定する整数です。いったんこの大きさに達したら、挿入はキューの要素が消費されるまでブロックされます。もし maxsize が0以下であるならば、キューの大きさは無限です。

最小の値を持つ要素から取り出されます (最小の要素とは、min(entries) が返す要素です)。 典型的な要素の形式は (priority_number, data) というタプルです。

data 要素を比較できない場合は、データ自体を無視して優先度の数値だけを比較するクラスで包めます。

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)
class queue.SimpleQueue

サイズの上限がない FIFO キューのコンストラクタです。 単純なキューには、タスクの追跡などの高度な機能はありません。

単純なキューは、要素の型を指定できる ジェネリック型 です。

Added in version 3.7.

exception queue.Empty

空の Queue オブジェクトで、非ブロックメソッド get() (または get_nowait()) が呼ばれたとき、送出される例外です。

exception queue.Full

満杯の Queue オブジェクトで、非ブロックメソッド put() (または put_nowait()) が呼ばれたとき、送出される例外です。

exception queue.ShutDown

シャットダウン済みの Queue オブジェクトで put() または get() を呼び出すと送出される例外です。

Added in version 3.13.

キューオブジェクト

キューオブジェクト(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=True, timeout=None)

item をキューに入れます。 もしオプション引数 block が真で timeout が None (デフォルト) の場合は、必要であればフリースロットが利用可能になるまでブロックします。 timeout が正の数の場合は、最大で timeout 秒間ブロックし、その時間内に空きスロットが利用可能にならなければ、例外 Full を送出します。 そうでない場合 (block が偽) は、空きスロットが直ちに利用できるならば、キューにアイテムを置きます。 できないならば、例外 Full を送出します (この場合 timeout は無視されます)。

キューがシャットダウン済みの場合は ShutDown を送出します。

Queue.put_nowait(item)

put(item, block=False) と等価です。

Queue.get(block=True, timeout=None)

キューからアイテムを取り除き、それを返します。 オプション引数 block が真で timeout が None (デフォルト) の場合は、必要であればアイテムが取り出せるようになるまでブロックします。 もし timeout が正の数の場合は、最大で timeout 秒間ブロックし、その時間内でアイテムが取り出せるようにならなければ、例外 Empty を送出します。 そうでない場合 (block が偽) は、直ちにアイテムが取り出せるならば、それを返します。 できないならば、例外 Empty を送出します (この場合 timeout は無視されます)。

Prior to 3.0 on POSIX systems, and for all versions on Windows, if block is true and timeout is None, this operation goes into an uninterruptible wait on an underlying lock. This means that no exceptions can occur, and in particular a SIGINT will not trigger a KeyboardInterrupt.

キューがシャットダウン済みで空の場合、または即時シャットダウン済みの場合は ShutDown を送出します。

Queue.get_nowait()

get(False) と等価です。

キューに入れられたタスクが全てコンシューマスレッドに処理されたかどうかを追跡するために 2つのメソッドが提供されます。

Queue.task_done()

過去にキューに入れられたタスクが完了した事を示します。キューのコンシューマスレッドに利用されます。タスクの取り出しに使われた各 get() の後に task_done() を呼び出すと、取り出したタスクに対する処理が完了した事をキューに教えます。

join() がブロックされていた場合、全itemが処理された (キューに put() された全てのitemに対して task_done() が呼び出されたことを意味します) 時に復帰します。

キューに追加された要素の総数を超えて呼び出すと ValueError を送出します。

Queue.join()

キューにあるすべてのアイテムが取り出されて処理されるまでブロックします。

未完了のタスクのカウント値はキューにアイテムが追加されるときは常に加算され、コンシューマースレッドが task_done() を呼び出してアイテムの回収とその全処理の完了が示されるときは常に減算されます。未完了のタスクのカウント値がゼロになった場合、join() のブロックが解除されます。

タスクの完了を待つ

キューに入れたタスクが完了するのを待つ例:

import threading
import queue

q = queue.Queue()

def worker():
    while True:
        item = q.get()
        print(f'Working on {item}')
        print(f'Finished {item}')
        q.task_done()

# ワーカースレッドを起動する。
threading.Thread(target=worker, daemon=True).start()

# ワーカーに 30 件のタスクを送る。
for item in range(30):
    q.put(item)

# すべてのタスクが完了するまでブロックする。
q.join()
print('All work completed')

キューを終了する

Queue オブジェクトが不要になったら、空になるまで処理を進めて終了することも、強制シャットダウンで即座に終了することもできます。

Queue.shutdown(immediate=False)

Queue インスタンスをシャットダウンモードにします。

キューにはこれ以上要素を追加できなくなります。 以後の put() の呼び出しは ShutDown を送出します。 現在 put() でブロックしている呼び出し元はブロックを解除され、そのスレッドで ShutDown が送出されます。

immediate が偽 (デフォルト) の場合、get() で格納済みのタスクを取り出し、通常の手順で処理を進めてキューを終了できます。

残っている各タスクに対して task_done() を呼び出せば、待機中の join() のブロックは通常どおり解除されます。

キューが空になると、以後の get() の呼び出しは ShutDown を送出します。

immediate が真の場合、キューは直ちに終了します。 キューの内容はすべて取り除かれ、未完了タスク数は取り除いたタスクの数だけ減ります。 未完了タスク数がゼロになれば、join() の呼び出し元のブロックが解除されます。 get() の呼び出し元もブロックを解除され、キューが空なので ShutDown が送出されます。

immediate を真に設定するときは、join() の使用に注意してください。 タスクが処理されていなくても join() のブロックが解除され、キューの完了待機で通常成り立つ不変条件が破られます。

Added in version 3.13.

SimpleQueue オブジェクト

SimpleQueue オブジェクトは以下のpublicメソッドを提供しています。

SimpleQueue.qsize()

キューの近似サイズを返します。ここで、qsize() > 0 であるからといって、後続の get() の呼び出しがブロックしないことが保証されないことに注意してください。

SimpleQueue.empty()

キューが空の場合は True を返し、そうでなければ False を返します。 empty() が False を返しても、後続の get() の呼び出しがブロックしないことは保証されません。

SimpleQueue.put(item, block=True, timeout=None)

item をキューに入れます。 このメソッドは決してブロックせず、常に成功します (メモリ割り当ての失敗など、低水準のエラーが起こる可能性はあります)。 省略可能な引数 block と timeout は無視され、Queue.put() との互換性のためだけに用意されています。

このメソッドには再入可能な C 実装があります。 つまり、同じスレッド内の別の put() 呼び出しが put() または get() の呼び出しに割り込んでも、デッドロックやキュー内部の状態の破損が起こりません。 そのため、__del__ メソッドなどのデストラクタや weakref のコールバックで使うのに適しています。

SimpleQueue.put_nowait(item)

put(item, block=False) と等価です。Queue.put_nowait() との互換性のためのメソッドです。

SimpleQueue.get(block=True, timeout=None)

キューからアイテムを取り除き、それを返します。 オプション引数 block が真で timeout が None (デフォルト) の場合は、必要であればアイテムが取り出せるようになるまでブロックします。 もし timeout が正の数の場合は、最大で timeout 秒間ブロックし、その時間内でアイテムが取り出せるようにならなければ、例外 Empty を送出します。 そうでない場合 (block が偽) は、直ちにアイテムが取り出せるならば、それを返します。 できないならば、例外 Empty を送出します (この場合 timeout は無視されます)。

SimpleQueue.get_nowait()

get(False) と等価です。

参考

multiprocessing.Queue クラス

(マルチスレッドではなく) マルチプロセスの文脈で使用されるキュークラス。

collections.deque は、サイズの上限がないキューの別の実装です。 ロックを必要としない高速でアトミックな append() 操作と popleft() 操作を備え、インデックスによるアクセスもサポートしています。