如何在 RabbitMQ 中实现轮询主题交换答案

【问题标题】：How to achieve round-robin topic exchange in RabbitMQ如何在 RabbitMQ 中实现轮询主题交换
【发布时间】：2017-07-20 03:43:00
【问题描述】：

我知道在主题交换中实现循环行为可能很棘手或不可能，所以我的问题实际上是我是否可以从 RabbitMQ 中获得任何东西，或者转向支持它的其他消息队列。

下面是我的申请要求的详细说明：

会有一位制作人，姑且称之为P
（可能）会有成千上万的消费者，我们称他们为Cn
每个消费者可以“订阅”一个或多个主题交换，并且多个消费者可以订阅同一个主题
发布到主题的每条消息都应仅由一个消费者使用

用例 #1

假设：

主题

foo.bar
foo.baz

消费者

消费者C1订阅了主题#
消费者C2订阅了主题foo.*
消费者C3订阅了主题*.bar

生产者P 发布以下消息：

发布foo.qux:C1 和C2 可能会使用此消息，但只有一个人会收到它
发布 foo.bar: C1、C2 和 C3 可能会使用此消息，但只有一个人会收到它

注意不幸的是，我不能为每个“主题”设置单独的队列，因此使用 Direct Exchange 不起作用，因为主题组合的数量可能很大（数万）

据我所知，RabbitMQ 没有开箱即用的解决方案。有没有人知道解决方法或者有另一个消息队列解决方案可以支持这个，例如。 Kafka、Kinesis 等。

谢谢

【问题讨论】：

虽然我理解这个问题并且发现它在理论上非常有趣（这可以说是 100% 足以在 SO 上提问），但我也很好奇这个用例。在我看来，主题交换旨在发布关于“主题”的消息，并让“人们”对接收它的某种类型的消息感兴趣。看不出从主题交换中“生产/消费”有什么用处，但我一定是错的。就个人而言，除了使用某些数据存储来推迟获胜消费者的选举之外，我没有看到任何其他方法。我可能错了，我对 Kafka 和 Kinesis 了解不多。
我的观点是：如果必须通知所有（或部分）消费者已发送消息，则使用主题交换。如果只有一个消费者必须以生产者/消费者的方式消费消息，则使用直接或扇出交换。如果两者都需要，请同时使用！发布到两个交易所。但这并不能解决问题，事实上，因为你会有错误的消费者消费。
顺便说一句，虽然我不是 Kafka 专家，但我有 99% 的把握它在这里不会有帮助（甚至会更糟），因为它根本没有路由的概念。无论好坏，它都是 pub/sub，这意味着它甚至对生产者/消费者的东西都没有帮助。请注意，即使是 RabbitMQ 也不保证一次性交付。
@user1527491 不幸的是，我不能对实际的真实用例说太多（因为 NDA 和其他东西），但我已尽力解释这一点。我确实同意这是非常具有挑战性（且有趣）且有趣的事情要解决。我能想到的另一个例子是： > 想象多个彩票，每个彩票有一个或多个参赛者，每个参赛者可以参加一个或多个彩票，但每个彩票只有一个参赛者中奖者。
@user1527491 是的，经过几天的检查，似乎没有办法用纯 RabbitMQ 或任何其他消息队列来实现这一点，这非常可悲和令人失望。现在转移到事件驱动的数据库（RethinkDB）看看它提供了什么。

标签： rabbitmq apache-kafka message-queue amazon-kinesis

【解决方案1】：

似乎将交换机的角色（路由消息）和队列（为等待处理的消息提供存放位置）的角色混为一谈。将消息汇集到一个或多个队列是交换的工作，而将消息从队列汇集到多个消费者是队列的工作。循环赛只对后者起作用。

基本上，topic 交换通过复制消息进行操作，每个队列对应于与消息一起发布的主题。因此，任何对循环行为的期望都是错误的，因为它违背了topic 交换的定义。

所有这一切都是为了确定，根据定义，问题中提出的场景没有意义。这确实不意味着不可能实现所需的行为，但术语和拓扑可能需要进行一些明确的调整。

让我们退后一步，看看所描述的一条消息的生命周期：它仅由一个生产者生产，并由众多消费者之一消费。通常，这是直接交换解决的方案。复杂的因素是您的消费者对他们将消费的消息类型有选择性（或者，换句话说，您的生产者对于它产生的消息类型不一致）。

通常在面向消息的处理中，一个消息类型对应一个消费者类型。因此，每种不同类型的消息都会有自己对应的队列。但是，根据这个问题中给出的描述，一个消息类型可能对应多个不同的消费者类型。我遇到的一个问题是以下声明：

很遗憾，我不能为每个“主题”设置单独的队列

从表面上看，这种说法毫无意义，因为它真正说的是您有任意多（实际上是未知数量）的消息类型；如果是这样的话，那么你将如何编写代码来处理它们？

因此，暂时忽略该声明，我们会发现 RabbitMQ 开箱即用的两种可能性：

使用直接交换并使用消息类型作为路由键发布您的消息。然后，让您的各种消费者只订阅他们可以处理的消息类型。 这是最常见的消息处理模式。
像你一样使用主题交换，并提出某种外部重复数据删除逻辑（可能是 memcached），在其中检查消息并在另一个消费者开始处理它时丢弃它。

现在，这些都没有明确处理循环要求。由于没有解释为什么或如何重要，因此假设可以忽略它。如果不是，则需要进一步定义问题空间。

【讨论】：

确实，我寻求使用 RabbitMQ 的方式并不是它的设计目的。但是，因为没有工具可以提供（至少据我所知），我试图改变消息代理的原则来涵盖我的用例。关于不幸的是，我不能为每个“主题”设置一个单独的队列：并不是消息不同，只有一种消息类型，而是针对不同的消费者群体，其中只有一个消费者是随机挑选的处理它。
嗯，有工具，但没有一个工具 - 但什么时候有呢？如果一切都是开箱即用的，那么对程序员的需求就会非常低。 :)
根据您评论的后半部分，听起来您将为每个消费者组创建一个队列，然后设置您的消费者以从适当的队列中提取。这是 RabbitMQ 中的默认功能。
目前我实现它的方法是使用 RabbitMQ 和数据存储来选择我希望向其发布消息的消费者。堆栈是 RabbitMQ -> Broker Consumer (does a db select) -> Deepstream (to push the message to the correct consumer for processing) 这并不理想，但它可以按照我想要的方式工作，至少现在是这样，直到我有更多时间深入研究 RabbitMQ/Kafka
我不清楚您的要求是什么，但很清楚的是消息传递可能不是正确的实现。如果您确实有更多时间，澄清需求（而不是实现）会有所帮助。