java - 使用 amqp 从队列中多路分解消息以在并行流中处理？

Question

我想弄清楚我是否可以从阻塞场景切换到更具 react 性的模式。

我有传入的更新命令到达队列，我需要按顺序处理它们，但只处理那些与同一实体有关的命令。本质上，只要没有两个流包含关于同一实体的事件，我就可以创建任意数量的并行更新事件流。

我在想，主队列的消费者可能能够利用 amqp 的路由机制和临时队列，通过为每个实体 ID 创建临时队列，并将消费者挂接到它们。一旦订阅者完成并且队列中当前没有关于所讨论实体的其他事件，队列就可以被处理掉。

这种情况是否经常使用？有没有更好的方法来实现这一目标？在我们当前的系统中，我们使用基于 id 的命名锁来防止并发更新。

Answer 1

至少有2个选项:

每个实体一个队列一个 Entity-Queue 上的 n 个消费者。

一个队列包含所有实体的消息。消息包含数据的地方它对实体来说是什么。你可以将它分成几个队列(一个 AMQP 队列用于一种类型的实体)或使用 BlockingQueue实现。

在 qmqp-queues 中拆分实体的好处

• 您可以使用 rabbitmq 创建 ha-setup
• 您可以路由消息
• 你可能有一个实体队列的多个消费者，如果有朝一日是必要的(可扩展性)
• 消息可以是持久的，因此可以在应用程序崩溃

使用内部 BlockingQueue 实现的好处

• 速度更快(显然没有 net-io)
• 一切都必须在一个 JVM 中发生

无论如何，这取决于您想要什么，因为这两种方式各有好处。

更新:我不确定我现在是否找到你，但让我给你一些资源来尝试一些事情。有专门的rabbitmq extensions也许其中一些可以给你一个想法。看看lternate exchanges和 exchange to exchange绑定(bind)。

同样对于基本测试，我不确定它是否涵盖所有 rabbitmq 功能或所有 amqp 功能，但 this有时会很有用。请记住，此可视化中的路由键是生产者名称，您还可以在那里找到一些示例。导入和导出您的配置。