rabbitmq消息发送的可靠性机制

淩亂°似流年 2023-08-17 16:49 149阅读 0赞

1.消息发送时确认送达exchange, 可以使用事物机制,或者是消息确认机制

2.消息送达exchange后, 设置mandatory为true,确保消息从exchange送达queue

3.exchange, queue, 消息都要做持久化

4.确保消息被消费者成功处理, 消费者需要设置autoAck为false(如果为true,只要消费端不抛出异常则被ack, 如果消费端接受消息后宕掉, 则该消息也会被视为成功处理)

以上四条确保了消息"至少一次"被投递, 其中一二条是关于消息发送中的可靠性

每个rabbitTemplate只能有一个confirm-callback和return-callback, 如果同一个工程中需要多个配置, 如有的消息需要确认送达, 有的不需要,则需要实例化多个amqpTemplate(rabbitTemplate)

关于一: 若使用confirm-callback,必须要配置publisherConfirms为true

//确认消息是否到达broker服务器，也就是只确认是否正确到达exchange中即可，只要正确的到达exchange中，broker即可确认该消息返回给客户端ack。

rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){
         @Override
         public void confirm(CorrelationData correlationData, boolean ack, String cause) {
             if (ack) {
                 System.out.println("消息确认成功");
             } else {
                 //处理丢失的消息（nack）
                System.out.println("消息确认失败");
             }
         }
    });

关于二: 使用return-callback时必须设置mandatory为true, 设置publisherReturns也为true

rabbitTemplate.setMandatory(true);  
//确认消息是否到达broker服务器，也就是只确认是否正确到达exchange中即可，只要正确的到达exchange中，broker即可确认该消息返回给客户端ack。

rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
         @Override
         public void returnedMessage(Message message, int replyCode, String replyText,
                                     String exchange, String routingKey) {
             //重新发布
             RepublishMessageRecoverer recoverer = new RepublishMessageRecoverer(errorTemplate,"errorExchange", "errorRoutingKey");
             Throwable cause = new Exception(new Exception("route_fail_and_republish"));
             recoverer.recover(message,cause);
             System.out.println("Returned Message："+replyText);
         }
    });