concurrency - 您如何在确保每个实体的 FIFO 的同时并行处理消息？

Question

假设您的系统中有一个实体，例如“Person”，并且您想要处理修改各种 Person 实体的事件。重要的是:

同一个人的事件按 FIFO 顺序处理

多个 Person 事件流由不同的线程/进程并行处理

我们有一个使用共享数据库和锁来解决这个问题的实现。线程竞争获取Person 的锁，然后在获取锁后按顺序处理事件。我们想转移到消息队列以避免轮询和锁定，我们认为这会减少数据库的负载并简化消费者代码的实现。

我已经对 ActiveMQ、RabbitMQ 和 HornetQ 进行了一些研究，但我没有看到实现这一点的明显方法。

ActiveMQ 支持消费者订阅通配符，但我没有看到将每个队列上的并发限制为 1 的方法。如果我可以这样做，那么解决方案将很简单:

以某种方式告诉代理允许所有以/queue/person 开头的队列的并发性为 1。

发布者使用队列名称中的人员 ID 将事件写入队列。例如:/queue/person.20

消费者使用通配符订阅队列:/queue/person。>

每个消费者将接收不同人员队列的消息。如果所有的人队列都在使用中，那么一些消费者可能会闲着，这没关系

处理完一条消息后，消费者发送一个 ACK​​，它告诉代理它已经完成了该消息，并允许该 Person 队列的另一条消息发送到另一个消费者(可能是同一个)

ActiveMQ 接近:您可以进行通配符订阅并启用“独占消费者”，但这种组合会导致单个消费者接收发送到所有匹配队列的所有消息，从而将所有人员的并发性降低到 1。我觉得我错过了一些明显的东西。

问题:

有没有办法用任何主要的消息队列实现来实现上述方法？我们对选择持相当开放的态度。唯一的要求是它在 Linux 上运行。

有没有不同的方法来解决我没有考虑的一般问题？

谢谢!

Answer 1

看起来 JMSXGroupID 正是我要找的。来自 ActiveMQ 文档:

http://activemq.apache.org/message-groups.html

他们的股票价格示例用例正是我所追求的。我唯一关心的是如果单个消费者死亡会发生什么。希望代理会检测到并选择另一个消费者与该组 ID 关联。