使用 RabbitMq 锁定和批量获取消息

Question

我正在尝试以更非传统的方式使用 RabbitMq（尽管此时我可以根据需要选择任何其他消息队列实现）。消费者没有将 Rabbit 推送消息留给我的消费者，而是连接到一个队列并获取一批 N 条消息（在此期间它消耗一些消息并可能拒绝一些消息），然后跳转到另一个队列等等。这样做是为了冗余。如果某些消费者崩溃，则所有消息都保证会被其他消费者消费。

问题是我有多个消费者，我不希望他们在同一个队列中竞争。有没有办法保证对队列的锁定？如果不是，我是否至少可以确保如果 2 个消费者连接到同一个队列，他们不会读取相同的消息？事务可能在某种程度上对我有帮助，但我听说它们将从 RabbitMQ 中删除。

也欢迎其他架构建议。

谢谢！

编辑： 正如评论中所指出的，我需要如何处理消息有一个特殊性。它们仅在分组中才有意义，并且相关消息很可能在队列中聚集在一起。例如，如果我提取了一批 100 条消息，那么我很有可能能够对消息 1-3、4-5,6-10 等做一些事情。如果我无法为某些消息找到一个组，我'将它们重新提交到队列中。 WorkQueue 不起作用，因为它会将消息从同一组传播到多个不知道如何处理它们的工作人员。

Answer 1

您可以设置通道/消费者级别的预取计数以批量消费消息。为了重新提交消息，您应该使用 basic.reject AMQP 方法，并且可以选择将这些消息重新排队或转发到死信队列。多个消费者尝试从同一个队列中提取消息不是问题，因为 AMQP basic.get 方法将同步以处理并发消费者。

https://groups.google.com/forum/#!topic/rabbitmq-users/hJ8f5du-GCA

Answer 2

您在Enterprise Integration Patterns 上看过这本免费的在线书籍吗？

听起来您确实需要一个工作流程，其中在消息到达您的工作人员之前有一个批处理器组件。使用 RabbitMQ 有两种方法可以做到这一点。要么使用可以为您进行批处理的交换类型（和消息格式），要么使用一个队列和一个对批次进行分类并将每个批次放在自己的队列中的工作人员。批处理器可能还应该向控制队列发送“批处理就绪”消息，以便工作人员可以发现新批处理队列的存在。批处理完成后，工作人员可以删除批处理队列。

如果您可以控制消息格式，则可以让 RabbitMQ 以几种方式隐式执行批处理。通过主题交换，您可以确保每条消息上的路由键的格式为 work.batchid.something，然后得知存在批处理 xxyzz 的工作人员将使用 #.xxyzz.# 之类的绑定键。消费这些消息。无需重新发布。

另一种方法是在标头中包含批次 ID，并使用较新的标头交换类型。当然，如果您愿意编写少量 Erlang 代码，您也可以实现自己的自定义交换类型。

我确实建议您查看这本书，因为它比大多数人开始使用的典型工作队列概念提供了更好的消息传递架构概述。

Answer 3

让您的消费者只从一个队列中拉出。它们将被保证不共享消息（Rabbit 将在当前连接的消费者之间循环消息），并且针对该确切的使用模式进行了大量优化。

开箱即用。在 RabbitMQ 文档中，它被称为 Work Queue 模型。一个队列，多个消费者，他们都不共享任何东西。这听起来像你需要的。