python - 删除 RabbitMQ (AMQP) 服务器上队列中的消息

Question

我有 1 个大任务，其中包含 200 个子任务(消息)，这些子任务将发布到队列中。如果我想取消这 1 个任务，则应删除 200 条消息(或剩余未处理的消息)。有什么方法可以删除队列中这些已发布的消息吗？

我能想到的一个解决方案是创建一个队列 (Q)，我在其中发布新队列 (X) 的名称。然后每个消费者连接到这个新动态创建的队列 (X) 并处理 200 条发布的消息。如果我想中止整个任务，我只从发布者端删除该队列 (X)。这是一种常见的方法吗？

Answer 1

我发现您建议的方法几乎没有问题。

第一个问题是由于 RMQ consumer prefetch旨在通过减少对代理的请求量来提高性能。如果您的消费者检索了一批任务，他们将在请求新任务之前处理所有任务，只有到那时他们才会意识到队列已被取消。因此，您的取消请求在大多数情况下都不会得到妥善处理。您可以将预取计数减少到 1 以避免这种副作用，但这会增加网络压力并降低整体速度。

第二个问题是AMQP协议(protocol)没有提供优雅处理队列删除的机制。因此，您的消费者需要小心处理队列消失，否则它们会崩溃。通过这样做，您将失去对错误和问题的可见性。您如何区分队列何时被显式删除与队列实际崩溃的情况？

在这种情况下，我建议的是使用父作业的标识符标记所有任务。每次消费者开始使用新任务时，它都会检查父作业是否有效或是否已被取消。在后一种情况下，它会简单地忽略任务并转到下一个任务。您需要为此提供支持服务。例如，一个 Redis 实例就足够了。

这种机制会更加简单和健壮。您可以根据需要旋转任意数量的消费者，而无需协调他们与正确队列的连接。无序或交错的任务也不成问题。