根据自定义属性过滤/删除 ActiveMQ 队列中的重复消息

Question

问题

当我的 Web 应用程序更新数据库中的项目时，它会通过 Camel 将包含项目 ID 的消息发送到 ActiveMQ 队列，该队列的使用者将获得更新的外部服务 (Solr)。外部服务独立读取数据库。

我想要的是，如果 Web 应用程序发送另一条具有相同项目 ID 的消息，而旧消息仍在队列中，则删除新消息以避免运行 Solr 更新两次。

在处理完更新请求并且具有该项目 ID 的消息离开队列后，应再次接受具有相同 ID 的新请求。

有没有办法让这项工作开箱即用？我真的很想放弃 ActiveMQ 并简单地将更新请求队列实现为具有唯一约束的数据库表，按时间戳或正在运行的插入 ID 排序。

到目前为止我尝试了什么

我已阅读 Stackoverflow 上的 this 和 this 页面。这些是那里提到的解决方案：

• Camel 中的幂等消费者：在这里，我可以指定一个表达式来定义构成重复的内容，但这也会阻止所有未来发送相同消息的尝试，即更新相同的项目。我只想在新的更新请求仍在队列中时丢弃它们。

• “ActiveMQ 已经进行重复检查，看看auditDepth！”：嗯，这看起来是一个好的开始，并且肯定最接近我想要的，但这决定了基于我无法设置的消息 ID。所以或者我找到了一种方法让 ActiveMQ 以某种方式为这个队列生成消息 ID或者我找到了一种方法来让审计内容查看我的项目 ID 字段而不是消息 ID。 （一个comment in my second link 甚至建议使用“您在标题上设置的定义明确的属性”，但未能解释如何。）

• 编写一个自定义插件，如果传入消息与队列中已经存在的消息匹配，则将它们重定向到死信队列。这似乎是迄今为止提供的最完整的解决方案，但对于我认为是相当平凡的日常任务来说，它感觉太过分了。

PS：我发现another SO page 提出同样的问题而没有答案。

Answer 1

您想要的不是消息代理功能，请在我之后重复，“消息代理不是数据库，消息代理不是数据库”，根据需要重复。

代理的工作是可靠地从 A 点到 B 点获取消息。客户端通过消息选择器提供一些过滤功能，但这是最小的，主要用于仅保留单个客户端感兴趣的特定消息从那里流动而不是其他一些客户可能负责处理的其他内容。

您的用例需要一个更有状态的以数据库为中心的解决方案，正如您所描述的那样。创建一个代理插件来遍历队列以检查消息是在重新发明轮子，如果队列深度很大，则容易出错，因为 ActiveMQ 甚至可能不会根据内存限制为您分页所有消息。