Python中asyncio.Queue异步队列的实现

import asyncio

queue = asyncio.Queue(maxsize=10)  # 有界队列，容量10
queue = asyncio.Queue()            # 无界队列（maxsize=0）

class Queue(mixins._LoopBoundMixin):
    def __init__(self, maxsize=0):
        self._maxsize = maxsize
        self._getters = collections.deque()   # 等待获取的 Future 队列
        self._putters = collections.deque()   # 等待放入的 Future 队列
        self._unfinished_tasks = 0            # 未完成任务计数
        self._finished = locks.Event()        # join() 阻塞用的事件
        self._finished.set()
        self._init(maxsize)
        self._is_shutdown = False

# put() 核心逻辑（简化）
async def put(self, item):
    while self.full():
        putter = self._get_loop().create_future()
        self._putters.append(putter)
        await putter                  # 协程在此挂起
    self.put_nowait(item)

# get() 核心逻辑（简化）
async def get(self):
    while self.empty():
        getter = self._get_loop().create_future()
        self._getters.append(getter)
        await getter                  # 协程在此挂起
    return self.get_nowait()

def _wakeup_next(self, waiters):
    while waiters:
        waiter = waiters.popleft()
        if not waiter.done():
            waiter.set_result(None)   # 唤醒一个等待的协程
            break

asyncio.Queue(maxsize=0)

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

queue.shutdown()              # 优雅关闭：允许消费完剩余元素
queue.shutdown(immediate=True) # 立即关闭：清空队列，中断所有等待

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}] 生产: {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}] 消费: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.8))
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=3)  # 有界队列，产生背压

    # 启动生产者和消费者
    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)]

    # 等待所有生产完成
    await asyncio.gather(*producers)
    # 等待队列中所有任务被消费完
    await queue.join()

    # 取消消费者（它们在 await queue.get() 处无限等待）
    for c in consumers:
        c.cancel()

asyncio.run(main())

try:
    item = await asyncio.wait_for(queue.get(), timeout=5.0)
except asyncio.TimeoutError:
    print("等待超时，队列中没有新数据")

try:
    await asyncio.wait_for(queue.put(item), timeout=3.0)
except asyncio.TimeoutError:
    print("队列已满，放入超时")

# 从其他线程安全地放入数据
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()
    # ... 启动 workers ...

    # 生产完毕后，优雅关闭队列
    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()

# 向每个 consumer 发送一个哨兵
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: 多个 fetcher 往 raw_queue 放数据
    fetchers = [asyncio.create_task(fetch(raw_queue)) for _ in range(5)]
    # Stage 2: 多个 processor 从 raw_queue 取出，处理后放入 processed_queue
    processors = [asyncio.create_task(process(raw_queue, processed_queue)) for _ in range(3)]
    # Stage 3: 单个 writer 从 processed_queue 取出并写入
    writer = asyncio.create_task(write(processed_queue))