java - 添加到 Activemq 队列时如何识别重复消息

Question

我正在 Spring Boot 应用程序中使用 ActiveMq 设置 JMS。但无法理解如何阻止发送者(消息提供者)向队列添加重复的消息。

我的应用程序初始 JMS 和消息转换器 Bean 配置如下

@Bean
    public Queue queue() {
        return new ActiveMQQueue("pendingDocuments.queue");
    }


@Bean 
 public MessageConverter jacksonJmsMessageConverter() {
       MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
            converter.setTargetType(MessageType.TEXT);
            converter.setTypeIdPropertyName("_type");
            return converter;
        }

发送方方法实现

@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;

@Autowired
private Queue pendingDocumentsQueue;

public void getPendingDocuments(){

/*My Custom Java Object, Actually these objects will read from DB for every 5 min, so process no way to know either these are already added to queue or not*/

Document document= new Document();
document.setUniqueId("122212");
document.setContent("TEST CONTENT");
this.jmsMessagingTemplate.convertAndSend(this.pendingDocumentsQueue, document);

        }

我想知道如何根据文档唯一 ID 将消息(我的文档对象)添加到队列中。

Answer 1

如果我理解正确的话，即使在崩溃的情况下，您也试图阻止您的通信处理重复消息？

通常情况下，您不应该有重复项(ActiveMQ 不会“自行”创建这些重复项)，但如果您考虑可能的故障点(发送方崩溃、代理崩溃、消费者崩溃)，您会发现发送者和代理无法做出“重复”的决定，除非消费者告诉他们最后一条实际处理的消息。

这可能是通过从消费者到代理的内部 ACK(防止丢失消息但不重复)或通过使用也可以防止重复的 JMS 事务来发生的。如果您让发送者和使用者使用 jmsTemplate.setSessionTransacted(true); (see here) 进行事务处理 session 您将使用如下协议(protocol)进行重复的免费通信:

1. 发送者从持久存储(可能是数据库)检查下一步是什么 唯一ID是。
2. 发件人发送消息。
3. 发送方同时提交交易 time 记录下处理此 UniqueID 的过程。
4. 经纪商将使用唯一的交易 ID 传递此消息。
5. 消费者收到消息。
6. 消费者处理这个UniqueID并同时提交 交易。