为什么 asyncio 队列 await get() 阻塞？

Question

为什么await queue.get()阻塞？

import asyncio

async def producer(queue, item):
    await queue.put(item)

async def consumer(queue):
    val = await queue.get()
    print("val = %d" % val)

async def main():
    queue = asyncio.Queue()
    await consumer(queue)
    await producer(queue, 1)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

如果我在consumer ()之前调用producer()，它工作正常也就是说，以下工作正常。

async def main():
    queue = asyncio.Queue()
    await producer(queue, 1)
    await consumer(queue)

为什么不等待 queue.get() 将控制权交还给事件循环，以便生产者协程可以运行，这将填充队列，以便 queue.get() 可以返回。

Answer 1

您需要并行启动消费者和生产者，例如main像这样定义：

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(consumer(queue), producer(queue, 1))

如果由于某种原因你不能使用gather，那么你可以这样做（相当于）：

async def main():
    queue = asyncio.Queue()
    asyncio.create_task(consumer(queue))
    asyncio.create_task(producer(queue, 1))
    await asyncio.sleep(100)  # what your program actually does

为什么不将await queue.get()控制权交还给事件循环，以便生产者协程可以运行，这将填充队列以便queue.get()可以返回。

await queue.get() 正在将控制权交还给事件循环。但是await意味着wait，所以当你的main协程说 时await consumer(queue)，这意味着“一旦consumer(queue)完成就恢复我”。由于consumer(queue)它本身在等待某人生产某物，因此您有一个经典的死锁案例。

颠倒顺序只是因为您的生产者是一次性的，所以它会立即返回给调用者。如果您的生产者碰巧等待外部源（例如套接字），那么您也会在那里陷入僵局。并行启动他们避免了僵局无论怎么producer和consumer写入。