キュー
******

**ソースコード:** 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" を送出します。

      キューがシャットダウン済みで空の場合、"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" を送出します。


優先度付きのキュー
==================

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

   シャットダウン済みのキューに対して "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())
