キュー¶

ソースコード: Lib/asyncio/queues.py

asyncio キューは queue モジュールのクラス群と同じ形になるように設計されています。 asyncio キューはスレッドセーフではありませんが、それらは async/await コードから使われるために特別に設計されています。

asyncio キューのメソッドは timeout パラメータを持たないことに注意してください; タイムアウトを伴うキューを使った処理を行うには asyncio.wait_for() 関数を使ってください。

下記の使用例節も参照してください。

Queue¶

class asyncio.Queue(maxsize=0)¶

先入れ先出し (FIFO) キューです。

maxsize がゼロ以下の場合、キューは無限長になります。 0 より大きい整数の場合、キューが maxsize に達すると await put() は get() によってキューの要素が除去されるまでブロックします。

標準ライブラリにおけるスレッドベースの queue モジュールと異なり、キューのサイズは常に既知であり、 qsize() メソッドを呼び出すことによって取得することができます。

バージョン 3.10 で変更: loop パラメータが削除されました。

このクラスはスレッド安全ではありません。

maxsize¶: キューに追加できるアイテム数です。

empty()¶: キューが空ならば True を、そうでなければ False を返します。

full()¶

キューに要素が maxsize 個あれば True を返します。

maxsize=0 (デフォルト) で初期化したキューでは、full() が True を返すことはありません。

async get()¶

キューから要素を削除して返します。キューが空の場合項目が利用可能になるまで待機します。

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

get_nowait()¶

直ちに利用できる要素がある場合はその要素を返し、そうでなければ QueueEmpty を送出します。

Raises QueueShutDown if the queue has been shut down and is empty.

async join()¶

キューにある全ての要素が取得され、処理されるまでブロックします。

未完了のタスクのカウント値は、キューにアイテムが追加されるときは常に加算され、キューの要素を消費するコルーチンが要素を取り出し、処理を完了したことを通知するために task_done() を呼び出すと減算されます。未完了のタスクのカウント値がゼロになると、 join() のブロックは解除されます。

async put(item)¶

要素をキューに入力します。キューが満杯の場合、要素を追加する前に空きスロットが利用できるようになるまで待機します。

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

put_nowait(item)¶

ブロックせずにアイテムをキューに追加します。

直ちに利用できるスロットがない場合、QueueFull を送出します。

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

qsize()¶: キュー内の要素数を返します。

shutdown(immediate=False)¶

Put a Queue instance into a shutdown mode.

The queue can no longer grow. Future calls to put() raise QueueShutDown. Currently blocked callers of put() will be unblocked and will raise QueueShutDown in the formerly awaiting task.

If immediate is false (the default), the queue can be wound down normally with get() calls to extract tasks that have already been loaded.

And if task_done() is called for each remaining task, a pending join() will be unblocked normally.

Once the queue is empty, future calls to get() will raise QueueShutDown.

If immediate is true, the queue is terminated immediately. The queue is drained to be completely empty and the count of unfinished tasks is reduced by the number of tasks drained. If unfinished tasks is zero, callers of join() are unblocked. Also, blocked callers of get() are unblocked and will raise QueueShutDown because the queue is empty.

Use caution when using join() with immediate set to true. This unblocks the join even when no work has been done on the tasks, violating the usual invariant for joining a queue.

Added in version 3.13.

task_done()¶

キューに投入されていた作業項目の処理が完了したことを通知します。

キューから作業を取り出す側が使います。 get() で作業項目を取り出すたびに、処理後に task_done() を呼び出して、作業が完了したことをキューに伝えます。

join() が現在ブロック中だった場合、全アイテムが処理されたとき (put() でキューに追加された全アイテムの task_done() の呼び出しを受信したとき) に再開します。

呼び出し回数がキューに追加されたアイテムの総数を超えると ValueError を送出します。

優先度付きのキュー¶

class asyncio.PriorityQueue¶

Queue の一種で、優先度の値が小さい要素から取り出します。

項目は典型的には (priority_number, data) 形式のタプルです。

LIFO キュー¶

class asyncio.LifoQueue¶: Queue の変種で、最後に追加された項目を最初に取り出します (後入れ先出し、またはスタック)。

例外¶

exception asyncio.QueueEmpty¶: この例外は get_nowait() メソッドが空のキューに対して呼ばれたときに送出されます。

exception asyncio.QueueFull¶: サイズが maxsize に達したキューに対して put_nowait() メソッドが呼ばれたときに送出される例外です。

exception asyncio.QueueShutDown¶: Exception raised when put(), put_nowait(), get() or get_nowait() is called on a queue which has been shut down.

Added in version 3.13.

使用例¶

キューを使って、並行処理を行う複数のタスクにワークロードを分散させることができます:

import asyncio
import random
import time


async def worker(name, queue):
    while True:
        # キューから作業項目を取り出す。
        sleep_for = await queue.get()

        # sleep_for 秒スリープする。
        await asyncio.sleep(sleep_for)

        # 作業項目の処理が完了したことをキューに通知する。
        queue.task_done()

        print(f'{name} has slept for {sleep_for:.2f} seconds')


async def main():
    # 作業を格納するキューを作成する。
    queue = asyncio.Queue()

    # ランダムな待ち時間を生成してキューに追加する。
    total_sleep_time = 0
    for _ in range(20):
        sleep_for = random.uniform(0.05, 1.0)
        total_sleep_time += sleep_for
        queue.put_nowait(sleep_for)

    # キューを並行して処理する三つのワーカータスクを作成する。
    tasks = []
    for i in range(3):
        task = asyncio.create_task(worker(f'worker-{i}', queue))
        tasks.append(task)

    # キューの全項目の処理が完了するまで待つ。
    started_at = time.monotonic()
    await queue.join()
    total_slept_for = time.monotonic() - started_at

    # ワーカータスクをキャンセルする。
    for task in tasks:
        task.cancel()
    # すべてのワーカータスクのキャンセルが完了するまで待つ。
    await asyncio.gather(*tasks, return_exceptions=True)

    print('====')
    print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds')
    print(f'total expected sleep time: {total_sleep_time:.2f} seconds')


asyncio.run(main())

キュー¶

Queue¶

優先度付きのキュー¶

LIFO キュー¶

例外¶

使用例¶

目次

前のトピックへ

次のトピックへ

このページ