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

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

queue モジュールは、複数プロデューサ-複数コンシューマ(multi-producer, multi-consumer)キューを実装します。これは、複数のスレッドの間で情報を安全に交換しなければならないときのマルチスレッドプログラミングで特に有益です。このモジュールの Queue クラスは、必要なすべてのロックセマンティクスを実装しています。

The module implements three types of queue, which differ only in the order in which the entries are retrieved. In a FIFO queue, the first tasks added are the first retrieved. In a LIFO queue, the most recently added entry is the first retrieved (operating like a stack). With a priority queue, the entries are kept sorted (using the heapq module) and the lowest valued entry is retrieved first.

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

In addition, the module implements a "simple" FIFO queue type, SimpleQueue, whose specific implementation provides additional guarantees in exchange for the smaller functionality.

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以下であるならば、キューの大きさは無限です。

The lowest valued entries are retrieved first (the lowest valued entry is the one that would be returned by min(entries)). A typical pattern for entries is a tuple in the form: (priority_number, data).

If the data elements are not comparable, the data can be wrapped in a class that ignores the data item and only compares the priority number:

from dataclasses import dataclass, field
from typing import Any

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

Constructor for an unbounded FIFO queue. Simple queues lack advanced functionality such as task tracking.

バージョン 3.7 で追加.

exception queue.Empty

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

exception queue.Full

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

キューオブジェクト

キューオブジェクト(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)

Put item into the queue. If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Full exception if no free slot was available within that time. Otherwise (block is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (timeout is ignored in that case).

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.

Queue.get_nowait()

get(False) と等価です。

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

Queue.task_done()

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

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

キューにある要素より多く呼び出された場合 ValueError が発生します。

Queue.join()

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

The count of unfinished tasks goes up whenever an item is added to the queue. The count goes down whenever a consumer thread calls task_done() to indicate that the item was retrieved and all work on it is complete. When the count of unfinished tasks drops to zero, join() unblocks.

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

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()

# Turn-on the worker thread.
threading.Thread(target=worker, daemon=True).start()

# Send thirty task requests to the worker.
for item in range(30):
    q.put(item)

# Block until all tasks are done.
q.join()
print('All work completed')

SimpleQueue オブジェクト

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

SimpleQueue.qsize()

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

SimpleQueue.empty()

Return True if the queue is empty, False otherwise. If empty() returns False it doesn't guarantee that a subsequent call to get() will not block.

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

Put item into the queue. The method never blocks and always succeeds (except for potential low-level errors such as failure to allocate memory). The optional args block and timeout are ignored and only provided for compatibility with Queue.put().

CPython 実装の詳細: This method has a C implementation which is reentrant. That is, a put() or get() call can be interrupted by another put() call in the same thread without deadlocking or corrupting internal state inside the queue. This makes it appropriate for use in destructors such as __del__ methods or weakref callbacks.

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 is an alternative implementation of unbounded queues with fast atomic append() and popleft() operations that do not require locking and also support indexing.