python - wait queue.put(item) 到 asyncio.Queue 上似乎没有释放事件循环的控制

Question

在这个简单的生产者/消费者示例中，就好像 awaitqueue.put(item) 不会释放事件循环以允许消费者运行直到完成。这会导致生产者将其所有项目放入队列中，然后消费者才能将其取出。

这是预期的吗？

如果我遵循 awaitqueue.put(item) 和 await asyncio.sleep(0)，我会得到我正在寻找的结果。

生产者然后将 1 个项目放入队列中，然后消费者从队列中取出 1 个项目。

我在 Python 3.6.8 和 3.7.2 中得到相同的结果。

import asyncio

async def produce(queue, n):
    for x in range(1, n + 1):
        print('producing {}/{}'.format(x, n))
        item = str(x)
        await queue.put(item)
        # await asyncio.sleep(0)
    await queue.put(None)

async def consume(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print('consuming item {}...'.format(item))

loop = asyncio.get_event_loop()
queue = asyncio.Queue(loop=loop)
producer_coro = produce(queue, 10)
consumer_coro = consume(queue)
loop.run_until_complete(asyncio.gather(producer_coro, consumer_coro))
loop.close()

Answer 1

This results in the producer putting all it's items onto the queue and only then the consumer gets to take them off. Is that expected?

是的。问题是您的队列是无界的，因此将某些内容放入其中永远不会挂起生产者，因此永远不会屈服于其他协程。这同样适用于立即提供数据的所有等待，例如一个read at EOF .

如果生产者的循环包含另一个暂停源，例如等待实际输入(毕竟它必须从某个地方获取项目)，那么这将导致它暂停，并且问题不会立即引起注意。使用 asyncio.sleep(0) 强制挂起也可以，但它很脆弱，因为它依赖于单个挂起来运行使用者。情况可能并不总是如此，因为消费者本身可以等待队列以外的某些事件。

无界队列在某些情况下是有意义的，例如当队列预先填充了任务时，或者生产者的体系结构将项目数量限制在合理的数量时。但如果队列项是动态生成的，最好添加一个界限。该界限保证了生产者的背压，并确保它不会垄断事件循环。