Python asyncio.Queueとは？非同期処理でデータを受け渡す方法

asyncio.Queueは、非同期処理の間でデータを受け渡すキューです。

「生産する側」と「処理する側」を分けたいときに使います。

---

基本の形

import asyncio

async def worker(name, q, out):
    while True:
        item = await q.get()
        if item is None:
            q.task_done()
            break
        await asyncio.sleep　(0.02)
        out.append((name, item))
        q.task_done()

async def main():
    q = asyncio.Queue()
    out = []

    ws = [asyncio.create_task(worker(f"W{i}", q, out)) for i in range(2)]

    for i in range(6):
        await q.put(i)

    await q.join()

    for _ in ws:
        await q.put(None)
    await asyncio.gather(*ws)

    return out

print(asyncio.run(main()))

実行結果：

[('W0', 0), ('W1', 1), ('W0', 2), ('W1', 3), ('W0', 4), ('W1', 5)]

2つのワーカーが交互に処理しています。

流れはqueue.Queueと同じです。

• put()で入れる
• get()で取り出す
• join()で全件完了を待つ
• Noneを送ってワーカーを終了させる

---

すべてawaitが必要

await q.put(item)
item = await q.get()
await q.join()

これがqueue.Queueとの最大の違いです。

put()・get()・join()はコルーチンなので、awaitを忘れると動きません。

q.put(item)      # 誤り：何も起きない

実行結果：

RuntimeWarning: coroutine 'Queue.put' was never awaited

一方でtask_done()はコルーチンではありません。awaitを付けないでください。

メソッド	await
put() / get() / join()	必要
task_done() / qsize() / empty()	不要
put_nowait() / get_nowait()	不要

---

queue.Queueとは別物

import asyncio
import queue

print(asyncio.QueueEmpty is queue.Empty)
print(asyncio.QueueEmpty.__module__)

実行結果：

False
asyncio.queues

例外クラスも別です。asyncio.QueueEmpty・asyncio.QueueFullを使ってください。

用途もはっきり分かれます。

	queue.Queue	asyncio.Queue
相手	スレッド間	コルーチン間
待ち方	スレッドをブロック	制御を譲る
スレッド安全	あり	なし

asyncio.Queueはスレッドセーフではありません。別スレッドから触ってはいけません。

スレッド間で渡すなら「Python queueとは？スレッド間で安全にデータを渡す方法」を参照してください。

---

maxsizeで流量を制限する

async def main():
    q = asyncio.Queue(maxsize=1)
    await q.put(1)

    try:
        await asyncio.wait_for(q.put(2), timeout=0.2)
    except asyncio.TimeoutError:
        return "put がブロックした"

print(asyncio.run(main()))

実行結果：

put がブロックした

いっぱいのときはput()側が待たされます。

これは重要な安全装置です。データの取得が処理より速い場合、制限がなければキューが無限に伸びてメモリを食い潰します。

APIから大量に取得しながら処理する、といった構成ではmaxsizeを必ず設定してください。

---

待たずに操作する

async def main():
    q = asyncio.Queue()
    try:
        q.get_nowait()
    except asyncio.QueueEmpty:
        return "asyncio.QueueEmpty"

print(asyncio.run(main()))

実行結果：

asyncio.QueueEmpty

get_nowait()・put_nowait()は待ちません。その場で失敗するか成功するかです。

---

gatherとの使い分け

「決まった数のタスクを並行に実行する」だけなら、キューは不要です。

results = await asyncio.gather(*[fetch(u) for u in urls])

キューが必要になるのは次のような場合です。

• 処理中に新しい仕事が増える（クロールで新しいURLが見つかる）
• ワーカーを常駐させたい
• 生産と消費の速度差を吸収したい
• 同時実行数をワーカー数で制御したい

単に同時実行数を絞るだけなら、Semaphoreのほうが簡潔です。

「Python asyncio.Semaphoreとは？同時実行数を制限する方法」を参照してください。

---

asyncio.Queueのまとめ

• コルーチン間でデータを受け渡す
• put()・get()・join()はawaitが必要
• task_done()にはawaitを付けない
• queue.Queueとは別物——スレッドセーフではない
• 例外はasyncio.QueueEmpty・asyncio.QueueFull
• maxsizeでメモリの暴走を防ぐ

---

あわせて読みたい

• Python queueとは？スレッド間で安全にデータを渡す方法
• Python asyncio.Semaphoreとは？同時実行数を制限する方法
• Python asyncio.create_task()とは？処理をすぐ走らせる方法
• Python Queueのtask_done()とjoin()とは？全タスク完了を待つ方法
• Python asyncio.gather()とは？複数の処理を同時に実行する方法
• Python asyncio クラス一覧｜Task・Future・Queue・Eventを解説