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

Question

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

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

我认为一个粗略的方法是。。。

在机器1上：

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

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

下面是我用这种方法很快注意到的一个陷阱。

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

为了解决这个问题，我需要在第3步中设置retry，直到消息成功发布到broker。

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

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

Answer 1

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

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

ActiveMQ Artemis确实内置了对检测从客户端发送的重复的支持。您可以在留档中阅读有关重复检测的更多信息。由于代理本身支持此行为，因此它提供了干净的配置、管理和监控。

在这两种情况下，您都需要在每条消息上设置一个特殊的标题，并使用“一个唯一的密钥来标识流程”，就像您为潜在的数据库解决方案所做的那样。此外，使用代理作为复制检测器总体上要简单得多。

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