import asyncio

queue = asyncio.Queue(maxsize=10)  # 有界隊(duì)列，容量10
queue = asyncio.Queue()            # 無(wú)界隊(duì)列（maxsize=0）

class Queue(mixins._LoopBoundMixin):
    def __init__(self, maxsize=0):
        self._maxsize = maxsize
        self._getters = collections.deque()   # 等待獲取的 Future 隊(duì)列
        self._putters = collections.deque()   # 等待放入的 Future 隊(duì)列
        self._unfinished_tasks = 0            # 未完成任務(wù)計(jì)數(shù)
        self._finished = locks.Event()        # join() 阻塞用的事件
        self._finished.set()
        self._init(maxsize)
        self._is_shutdown = False

# put() 核心邏輯（簡(jiǎn)化）
async def put(self, item):
    while self.full():
        putter = self._get_loop().create_future()
        self._putters.append(putter)
        await putter                  # 協(xié)程在此掛起
    self.put_nowait(item)

# get() 核心邏輯（簡(jiǎn)化）
async def get(self):
    while self.empty():
        getter = self._get_loop().create_future()
        self._getters.append(getter)
        await getter                  # 協(xié)程在此掛起
    return self.get_nowait()

def _wakeup_next(self, waiters):
    while waiters:
        waiter = waiters.popleft()
        if not waiter.done():
            waiter.set_result(None)   # 喚醒一個(gè)等待的協(xié)程
            break

asyncio.Queue(maxsize=0)

# 典型用法
await queue.join()   # 等待所有任務(wù)完成

queue.shutdown()              # 優(yōu)雅關(guān)閉：允許消費(fèi)完剩余元素
queue.shutdown(immediate=True) # 立即關(guān)閉：清空隊(duì)列，中斷所有等待

def _init(self, maxsize):
    self._queue = collections.deque()
def _put(self, item):
    self._queue.append(item)
def _get(self):
    return self._queue.popleft()

def _init(self, maxsize):
    self._queue = []
def _put(self, item, heappush=heapq.heappush):
    heappush(self._queue, item)
def _get(self, heappop=heapq.heappop):
    return heappop(self._queue)

def _init(self, maxsize):
    self._queue = []
def _put(self, item):
    self._queue.append(item)
def _get(self):
    return self._queue.pop()   # 從尾部取出

import asyncio
import random

async def producer(queue: asyncio.Queue, name: str):
    for i in range(5):
        item = f"{name}-item-{i}"
        await queue.put(item)
        print(f"[{name}] 生產(chǎn): {item}")
        await asyncio.sleep(random.uniform(0.1, 0.5))

async def consumer(queue: asyncio.Queue, name: str):
    while True:
        item = await queue.get()
        print(f"  [{name}] 消費(fèi): {item}")
        await asyncio.sleep(random.uniform(0.2, 0.8))
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=3)  # 有界隊(duì)列，產(chǎn)生背壓

    # 啟動(dòng)生產(chǎn)者和消費(fèi)者
    producers = [asyncio.create_task(producer(queue, f"P{i}")) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, f"C{i}")) for i in range(3)]

    # 等待所有生產(chǎn)完成
    await asyncio.gather(*producers)
    # 等待隊(duì)列中所有任務(wù)被消費(fèi)完
    await queue.join()

    # 取消消費(fèi)者（它們?cè)?await queue.get() 處無(wú)限等待）
    for c in consumers:
        c.cancel()

asyncio.run(main())

try:
    item = await asyncio.wait_for(queue.get(), timeout=5.0)
except asyncio.TimeoutError:
    print("等待超時(shí)，隊(duì)列中沒(méi)有新數(shù)據(jù)")

try:
    await asyncio.wait_for(queue.put(item), timeout=3.0)
except asyncio.TimeoutError:
    print("隊(duì)列已滿，放入超時(shí)")

# 從其他線程安全地放入數(shù)據(jù)
loop.call_soon_threadsafe(queue.put_nowait, item)

future = asyncio.run_coroutine_threadsafe(queue.put(item), loop)
future.result()  # 阻塞等待完成

async def main():
    queue = asyncio.Queue()
    # ... 啟動(dòng) workers ...

    # 生產(chǎn)完畢后，優(yōu)雅關(guān)閉隊(duì)列
    queue.shutdown()
    await queue.join()

SENTINEL = object()

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is SENTINEL:
            queue.task_done()
            break
        process(item)
        queue.task_done()

# 向每個(gè) consumer 發(fā)送一個(gè)哨兵
for _ in range(num_consumers):
    await queue.put(SENTINEL)
await queue.join()

async def pipeline():
    raw_queue = asyncio.Queue()
    processed_queue = asyncio.Queue()

    # Stage 1: 多個(gè) fetcher 往 raw_queue 放數(shù)據(jù)
    fetchers = [asyncio.create_task(fetch(raw_queue)) for _ in range(5)]
    # Stage 2: 多個(gè) processor 從 raw_queue 取出，處理后放入 processed_queue
    processors = [asyncio.create_task(process(raw_queue, processed_queue)) for _ in range(3)]
    # Stage 3: 單個(gè) writer 從 processed_queue 取出并寫入
    writer = asyncio.create_task(write(processed_queue))