python asyncio隊列示例

asyncio.Queue 是用於異步任務間安全通信的隊列工具，1. 生產者通過await queue.put(item) 添加數據，消費者用await queue.get() 獲取數據；2. 每處理完一項需調用queue.task_done()，以便queue.join() 等待所有任務完成；3. 使用None 作為結束信號通知消費者停止；4. 多個消費者時，需發送多個結束信號或在取消任務前確保所有任務已處理完畢；5. 隊列支持設置maxsize 限制容量，put 和get 操作自動掛起不阻塞事件循環，程序最終通過cancel() 終止仍在等待的消費者任務，完整實現高效異步生產者-消費者模式。

下面是一個實用的Python asyncio.Queue示例，展示如何在異步任務中使用隊列進行生產者-消費者模式的通信。

什麼是asyncio.Queue ？

asyncio.Queue是一個線程安全、異步安全的隊列，用於在多個協程之間安全地傳遞數據。常用於生產者生成數據、消費者處理數據的場景。

基本示例：生產者和消費者

import asyncio
import random

async def producer(queue, n):
    for i in range(n):
        item = f"item-{i}"
        await queue.put(item)
        print(f"Produced: {item}")
        # 模擬異步耗時（如網絡請求、IO）
        await asyncio.sleep(random.uniform(0.1, 0.5))
    # 發送結束信號await queue.put(None)
    print("Producer done.")

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            # 收到結束信號，把None 放回去，以便其他消費者也能收到queue.task_done()
            break
        print(f"Consumed: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.7)) # 模擬處理時間queue.task_done() # 標記任務完成print("Consumer done.")

async def main():
    queue = asyncio.Queue()
    num_items = 5

    # 同時運行生產者和消費者producer_task = asyncio.create_task(producer(queue, num_items))
    consumer_task = asyncio.create_task(consumer(queue))

    # 等待消費者處理完所有任務（包括None）
    await producer_task
    await queue.join() # 等待所有任務被標記為task_done
    consumer_task.cancel() # 手動取消消費者（因為它在等待更多任務）

# 運行asyncio.run(main())

關鍵點說明

✅ queue.put()和queue.get()

• 都是awaitable的，不會阻塞事件循環。
• put()在隊列滿時掛起； get()在隊列空時掛起。

✅ queue.task_done()

• 每當消費者處理完一個任務，必須調用task_done() 。
• queue.join()會等待所有任務都被task_done()標記。

✅ 使用None作為結束信號

• 通知消費者停止。
• 多個消費者時，要小心處理結束信號（避免多個消費者重複處理）。

✅ queue.join() vs task.cancel()

• queue.join()等待所有任務完成。
• 消費者是無限循環，需要用cancel()停止。

多個消費者示例（可選）

 async def main_multiple_consumers():
    queue = asyncio.Queue()
    num_items = 10
    num_consumers = 3

    # 創建多個消費者consumers = [asyncio.create_task(consumer(queue)) for _ in range(num_consumers)]
    producer_task = asyncio.create_task(producer(queue, num_items))

    await producer_task
    await queue.join() # 等待所有item 被處理# 取消所有消費者（它們在等待更多任務）
    for c in consumers:
        c.cancel()
    # 等待它們被取消await asyncio.gather(*consumers, return_exceptions=True)

# 運行# asyncio.run(main_multiple_consumers())

小貼士

• 隊列默認無大小限制。可以用asyncio.Queue(maxsize=5)限制容量。
• 如果隊列滿了， put()會自動等待，直到有空間。
• 不要用.get_nowait()或.put_nowait()在協程中，除非你確定不會出錯，否則會拋異常。

基本上就這些。 asyncio.Queue是寫異步爬蟲、任務調度、消息處理系統的常用工具，掌握它對寫高效的異步程序很有幫助。

以上是python asyncio隊列示例的詳細內容。更多資訊請關注PHP中文網其他相關文章！