asyncio.Queue() 的目的是什么以及如何正确使用它？

Question

我已经熟悉Python，但我刚刚开始学习“异步编程”和“io任务”主题。在教程中，我看到它asyncio.Queue()与生产者和消费者相关的东西一起使用，但我不太理解它。

有人可以简化一下，并举几个例子吗？

Answer 1

asyncio.Queues是普通 Python 的再现 queue，但具有特殊的异步属性。以下功能值得关注：

coroutine get()

从队列中删除并返回一个项目。如果队列为空，则等待直到有可用的项目。

coroutine put(item)

将一个项目放入队列中。如果队列已满，请等到有空闲槽位后再添加项目。

coroutine join()

阻塞直到队列中的所有项目都已被接收并处理。

如果您了解异步代码的一般思想（代码在同一线程中运行，但不断在操作之间切换），您应该会看到异步队列的一般用途 - 这是允许使用和方法进行灵活的await操作。get()put()

以收集、过滤和使用某些对象（无论您希望它们是什么）为例。假设您异步收集 30 个对象，其中每个收集函数将花费 1-5 秒。假设您想要过滤这些对象，过滤需要 4 到 7 秒。过滤功能在收集完成后立即启动，因此也许在某个时刻asyncio.loop您可能会进行以下操作：

object gathering...
object gathered -> filtering -> stored!
object gathering...
object gathering...
object gathered -> filtering...
object gathered -> filtering...
object gathering...
object gathered -> filtering -> stored!
...

其中 表示的对象-> filtering已经处于其操作的下一阶段，并且-> stored!已经被收集和过滤。当您意识到一旦达到存储的对象数量asyncio.queue时，最好停止过滤，因为过滤会占用线程并导致您无法有效地使用已过滤的对象，因此上述场景中的有用性就会出现。 。因此，你的堆栈最终可能看起来像这样：n

object gathering -> (waiting for queue to have space)
object gathered -> filtering -> stored!
object gathered -> (waiting for queue to have space)
object gathered -> (waiting for queue to have space)
object gathered -> filtered -> stored!
object gathered -> filtered -> stored!
object gathered -> (waiting for queue to have space)
object gathered -> filtering -> stored!
...

这是数以百万计的关于如何asyncio.queue使用的示例之一，但是上面示例的想法是让您开始了解异步编程如何成为一般编程的不同维度 - 您应该了解微观和微观事物的相关性。宏观尺度。函数如何自行运行，以及函数如何与其他函数交互（在异步函数的情况下，不允许另一个函数运行）。Asyncio 原语是这个概念的核心，非常值得理解。