キュー¶
ソースコード: 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を返します。
- async get()¶
キューから要素を削除して返します。キューが空の場合項目が利用可能になるまで待機します。
キューがシャットダウン済みで空の場合、または即時シャットダウンされた場合は、
QueueShutDownを送出します。
- get_nowait()¶
直ちに利用できる要素がある場合はその要素を返し、そうでなければ
QueueEmptyを送出します。キューがシャットダウン済みで空の場合、
QueueShutDownを送出します。
- async join()¶
キューにある全ての要素が取得され、処理されるまでブロックします。
未完了のタスクのカウント値は、キューにアイテムが追加されるときは常に加算され、キューの要素を消費するコルーチンが要素を取り出し、処理を完了したことを通知するために
task_done()を呼び出すと減算されます。未完了のタスクのカウント値がゼロになると、join()のブロックは解除されます。
- async put(item)¶
要素をキューに入力します。キューが満杯の場合、要素を追加する前に空きスロットが利用できるようになるまで待機します。
キューがシャットダウン済みの場合、
QueueShutDownを送出します。
- put_nowait(item)¶
ブロックせずにアイテムをキューに追加します。
直ちに利用できるスロットがない場合、
QueueFullを送出します。キューがシャットダウン済みの場合、
QueueShutDownを送出します。
- qsize()¶
キュー内の要素数を返します。
- shutdown(immediate=False)¶
Queueインスタンスをシャットダウンモードにします。以後、キューに要素を追加できなくなります。
put()を呼び出すとQueueShutDownを送出します。 すでにput()でブロックしている呼び出し元もブロックを解除され、待機していたタスク内でQueueShutDownが送出されます。immediate が偽 (デフォルト) の場合、
get()で投入済みのタスクを取り出し、通常どおり処理しながらキューの使用を終了できます。残りのタスクそれぞれについて
task_done()を呼び出せば、待機中のjoin()も通常どおりブロックを解除されます。キューが空になると、以後の
get()はQueueShutDownを送出します。immediate が真の場合、キューを即座に終了します。 キューからすべての要素を取り除いて空にし、取り除いたタスクの数だけ未完了タスク数を減らします。 未完了タスク数がゼロになると、
join()の呼び出し元のブロックを解除します。 また、get()でブロックしている呼び出し元もブロックを解除されますが、キューが空なのでQueueShutDownが送出されます。immediate を真にして
join()を使う場合は注意してください。 タスクを処理していなくてもjoin()のブロックが解除されるため、通常のキューの完了待ちで保証される条件が成り立たなくなります。Added in version 3.13.
- task_done()¶
キューに投入されていた作業項目の処理が完了したことを通知します。
キューから作業を取り出す側が使います。
get()で作業項目を取り出すたびに、処理後にtask_done()を呼び出して、作業が完了したことをキューに伝えます。join()が現在ブロック中だった場合、全アイテムが処理されたとき (put()でキューに追加された全アイテムのtask_done()の呼び出しを受信したとき) に再開します。呼び出し回数がキューに追加されたアイテムの総数を超えると
ValueErrorを送出します。
優先度付きのキュー¶
LIFO キュー¶
例外¶
- exception asyncio.QueueEmpty¶
この例外は
get_nowait()メソッドが空のキューに対して呼ばれたときに送出されます。
- exception asyncio.QueueFull¶
サイズが maxsize に達したキューに対して
put_nowait()メソッドが 呼ばれたときに送出される例外です。
- exception asyncio.QueueShutDown¶
シャットダウン済みのキューに対して
put()、put_nowait()、get()、get_nowait()を呼び出したときに送出される例外です。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())