如何让 ActiveMQ 检测到来自消息发布者（幂等生产者）的重复消息

Question

ActiveMQ 是否支持幂等生产者？我知道 Camel 有一个幂等消费者模式来检测和处理重复消息，但我想知道这是否可以在源（生产者）处防止。

这里有一点背景。我有水平扩展的应用程序访问相同的数据库。有一个特定的表维护特定进程的状态。这些水平应用程序应该能够读取状态并调用另一个进程，但是只有其中一个应该能够调用它。一旦满足所需条件，此应用程序会定期轮询数据库并将消息发布到消息传递代理。但我希望其中一个负载平衡应用程序应该能够发布消息。

我认为的一种粗略方法是......

在机器 1 上：

1. 读取数据库以检查是否满足必要条件。
2. 在向代理发布消息之前，将记录写入另一个状态表，并使用唯一键标识进程和提交。如果此操作由于违反唯一键约束而失败，则意味着另一台机器上的进程成功发布了消息。
3. 将消息发布到代理
4. 如果消息发布失败，由于某种原因，根据唯一键/主键对状态表进行删除操作。

在机器 2、3、4 等上运行的相同应用程序可以执行相同的操作。

以下是我很快注意到这种方法的一个缺陷。

假设机器 1 能够完成第 2 步，但未能执行第 3 步并继续执行第 4 步。同时，当第 2 步失败时，机器 2 将继续前进，而不再尝试再次读取状态并发布消息.

为了解决这个问题，我需要在第 3 步重试，直到消息成功发布到代理。

另一种选择是使用https://camel.apache.org/components/latest/eips/idempotentConsumer-eip.html 模式。但这本质上是消费者方面的过滤器。虽然这将达到我的目的，但在消息发布方面是否有类似的现成方法可用。

我想知道，这种方法是否正确或任何更好的替代方法，或者任何现有的库可以用于在本地或远程跨 JVM 执行锁定机制。

Answer 1

不清楚您使用的是哪个版本的 ActiveMQ（即ActiveMQ 5.x 或ActiveMQ Artemis），因此我将尝试解决这两个问题。

ActiveMQ 5.x 没有任何内置支持来检测从客户端发送的重复数据。但是，您可以使用broker plugin 来实现此功能。我在这里看到的唯一挑战是配置、管理和监控重复 ID 的缓存。

ActiveMQ Artemis 确实内置了对检测从客户端发送的重复项的支持。你可以read more about duplicate detection in the documentation。由于代理本身支持此行为，因此它提供了干净的配置、管理和监控。

在任何一种情况下，您都需要在每条消息上设置一个特殊的标头，其中包含“标识进程的唯一键”，就像您为潜在的数据库解决方案所做的那样。此外，使用代理作为重复检测器总体上更简单。

如果您当前使用的是 ActiveMQ 5.x，但想要迁移到 ActiveMQ Artemis 以使用重复检测功能，则您不一定需要更新客户端，因为 ActiveMQ Artemis 完全支持 5 使用的 OpenWire 协议。 x 客户。您应该能够将它们指向 ActiveMQ Artemis 的新实例，然后一切正常。