使用amqp将来自队列的消息多路分解为并行流处理？

Question

我正在尝试弄清楚是否可以从阻塞场景切换到更具反应性的模式。

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

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

这个场景是经常使用的吗？有没有更好的方法来实现这一点？在我们当前的系统中，我们使用基于id的命名锁来防止并发更新。

Answer 1

至少有两个选项：

每个实体队列的单个队列和一个实体队列上的n个消费者。

包含所有实体的消息的One队列。其中消息包含其对于实体的数据。然后，您可以将其拆分为几个队列(一种类型的实体对应一个AMQP-Queue )，也可以使用BlockingQueue实现。

拆分qmqp-queue中的实体的好处

• 您可以使用application-crash

创建ha-

• 您可以路由消息

• 如果有必要，您可能会有多个实体队列使用者某一天rabbitmq可能是持久性的，因此可以在rabbitmq上恢复

使用内部BlockingQueue实现的好处

net-io obviously)

• Everything的速度更快(-io
• 不必在一个JVM

中发生)。

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

更新：我不确定我现在是否理解你了，但让我给你一些资源来尝试一些东西。有特别的rabbitmq extensions，也许他们中的一些可以给你一个想法。看一下lternate exchanges和exchange to exchange绑定。

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