RabbitMQ：使用一致的哈希交换扩展队列答案

【问题标题】：RabbitMQ: Scaling queues with the consistent hash exchangeRabbitMQ：使用一致的哈希交换扩展队列
【发布时间】：2016-12-16 19:31:07
【问题描述】：

（从Github Issue接听）

我们使用 RabbitMQ 的 consistent hash exchange，这对于将路由密钥从交换分片到多个队列很有用。

我们使用队列将工作分配给工人。理想情况下，我们希望动态扩展这些工人，但这会带来问题。

要扩大规模，您需要添加一个新队列和绑定。就其本身而言，这并不是什么大问题 - 除了分片键现在可能开始进入不同队列的事实。

所以 Worker A 可能一直在管理 Thing1，但现在当我们添加新队列时，Worker B 可能最终会收到 Thing1 的消息。在 Worker B 开始接收 Thing1 消息之前，Worker A 必须完成所有 Thing1 处理，这一点很重要。

他们有什么方法或插件可以缓解这个问题吗？

【问题讨论】：

我的第一个问题是为什么不能为同一个队列使用多个工作人员？只有当你必须确保一定的顺序时，你才必须将一个工人限制在一个队列中。
我确实需要多次阅读。消息顺序对您很重要。因此，您需要在添加新工作人员之前检查所有队列是否为空。该插件声明当您添加新队列时存在竞争条件。在极少数情况下，您最终可能会收到重复的消息。在这种特殊情况下，我不会使用此插件添加所有内容。我将创建一个输入队列，其中一个调度员工作人员侦听，该进程负责将所有消息分发到正确的 msg 队列。
@slowjack2k 工作人员正在管理内存中的状态，因此一个工作人员只能管理 Thing1 的状态。
在这种情况下，我建议您使用调度员工作人员而不是哈希交换。调度员更灵活。它可以在为下一个请求使用新队列之前跟踪哪些消息已处理，哪些未处理。我认为消息重复的风险会对您的应用产生很多影响。

【解决方案1】：

在这种情况下，我将使用调度消息而不是散列交换的工作人员。

producer1 ... producern => topic-exchange => queue => dispatcher worker => queue1 ... queuen => worker1 ... worker

这样调度员工作人员可以跟踪所有消息。为此，您可以简单地检查队列中还剩下多少消息，或者您可以检查工作人员确认消息，或者您可以使用 rabbitmqs RPC 功能。

【讨论】：

所以我在一个单独的worker中有效地实现了分布式哈希交换（加上一些功能）？
在这种特殊情况下是的。它需要更多的工作，但可以为您提供所需的细粒度控制。我能想到的另一件事是一种控制器工作人员，您可以将其作为 rpc 实现“询问”，如果此时从实际工作人员中处理消息是有效的。
我没有问什么，也许你在你的生产者内部等待，直到这个生产者的最后一条消息被处理。那么你只需要改变你的制作人。
我可以改变任何我喜欢的东西，只是想知道最好的方法是什么，并希望一些 rabbitmq 插件可以解决我所有的问题:)。是的，我可以添加一个同步步骤 - 但这最终会减慢系统速度。我真正想要的是哈希交换能够记住它发送什么路由密钥的位置。
IMO 至少有 2 个原因说明此解决方案不是最佳的：1) 具有单个调度程序消费者及其单个队列，违背了将负载分片到多个队列的目的。 2）为了有效地实现这一点，您需要调度员和工作人员之间的某种临时通信。调度程序如何知道哪些工作人员和队列正在等待消息？调度员如何知道消费者是否意外停止？这一切都由交易所优雅地解决，而在调度程序中，您将不得不支付相当多的复杂性和开销