RabbitMQ中怎么保證消息的可靠投遞

今天就跟大家聊聊有關RabbitMQ中怎么保證消息的可靠投遞，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。

RabbitMQ整合Spring Boot，我們只需要增加對應的starter即可

<dependency>   <groupId>org.springframework.boot</groupId>   <artifactId>spring-boot-starter-amqp</artifactId> </dependency>

基于注解

在application.yaml的配置如下

spring:   rabbitmq:     host: myhost     port: 5672     username: guest     password: guest     virtual-host: /  log:   exchange: log.exchange   info:     queue: info.log.queue     binding-key: info.log.key   error:     queue: error.log.queue     binding-key: error.log.key   all:     queue: all.log.queue     binding-key: '*.log.key'

消費者代碼如下

@Slf4j @Component public class LogReceiverListener {      /**      * 接收info級別的日志      */     @RabbitListener(             bindings = @QueueBinding(                     value = @Queue(value = "${log.info.queue}", durable = "true"),                     exchange = @Exchange(value = "${log.exchange}", type = ExchangeTypes.TOPIC),                     key = "${log.info.binding-key}"             )     )     public void infoLog(Message message) {         String msg = new String(message.getBody());         log.info("infoLogQueue 收到的消息為: {}", msg);     }      /**      * 接收所有的日志      */     @RabbitListener(             bindings = @QueueBinding(                     value = @Queue(value = "${log.all.queue}", durable = "true"),                     exchange = @Exchange(value = "${log.exchange}", type = ExchangeTypes.TOPIC),                     key = "${log.all.binding-key}"             )     )     public void allLog(Message message) {         String msg = new String(message.getBody());         log.info("allLogQueue 收到的消息為: {}", msg);     } }

生產者如下

@RunWith(SpringRunner.class) @SpringBootTest public class MsgProducerTest {      @Autowired     private AmqpTemplate amqpTemplate;     @Value("${log.exchange}")     private String exchange;     @Value("${log.info.binding-key}")     private String routingKey;      @SneakyThrows     @Test     public void sendMsg() {         for (int i = 0; i < 5; i++) {             String message = "this is info message " + i;             amqpTemplate.convertAndSend(exchange, routingKey, message);         }          System.in.read();     } }

Spring Boot針對消息ack的方式和原生api針對消息ack的方式有點不同

原生api消息ack的方式

消息的確認方式有2種

自動確認(autoAck=true)

手動確認(autoAck=false)

消費者在消費消息的時候，可以指定autoAck參數

String basicConsume(String queue, boolean autoAck, Consumer callback)

autoAck=false: RabbitMQ會等待消費者顯示回復確認消息后才從內存(或者磁盤)中移出消息

autoAck=true: RabbitMQ會自動把發送出去的消息置為確認，然后從內存(或者磁盤)中刪除，而不管消費者是否真正的消費了這些消息

手動確認的方法如下，有2個參數

basicAck(long deliveryTag, boolean multiple)

deliveryTag: 用來標識信道中投遞的消息。RabbitMQ  推送消息給Consumer時，會附帶一個deliveryTag，以便Consumer可以在消息確認時告訴RabbitMQ到底是哪條消息被確認了。

RabbitMQ保證在每個信道中，每條消息的deliveryTag從1開始遞增

multiple=true: 消息id<=deliveryTag的消息，都會被確認

myltiple=false: 消息id=deliveryTag的消息，都會被確認

消息一直不確認會發生啥?

如果隊列中的消息發送到消費者后，消費者不對消息進行確認，那么消息會一直留在隊列中，直到確認才會刪除。

如果發送到A消費者的消息一直不確認，只有等到A消費者與rabbitmq的連接中斷，rabbitmq才會考慮將A消費者未確認的消息重新投遞給另一個消費者

Spring Boot中針對消息ack的方式

有三種方式，定義在AcknowledgeMode枚舉類中

spring boot針對消息默認的ack的方式為AUTO。

在實際場景中，我們一般都是手動ack。

application.yaml的配置改為如下

spring:   rabbitmq:     host: myhost     port: 5672     username: guest     password: guest     virtual-host: /     listener:       simple:         acknowledge-mode: manual # 手動ack，默認為auto

相應的消費者代碼改為

@Slf4j @Component public class LogListenerManual {      /**      * 接收info級別的日志      */     @RabbitListener(             bindings = @QueueBinding(                     value = @Queue(value = "${log.info.queue}", durable = "true"),                     exchange = @Exchange(value = "${log.exchange}", type = ExchangeTypes.TOPIC),                     key = "${log.info.binding-key}"             )     )     public void infoLog(Message message, Channel channel) throws Exception {         String msg = new String(message.getBody());         log.info("infoLogQueue 收到的消息為: {}", msg);         try {             // 這里寫各種業務邏輯             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);         } catch (Exception e) {             channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);         }     } }

我們上面用到的注解，作用如下

基于JavaConfig

既然用注解這么方便，為啥還需要JavaConfig的方式呢?

JavaConfig方便自定義各種屬性，比如同時配置多個virtual host等

具體代碼看GitHub把

RabbitMQ如何保證消息的可靠投遞

一個消息往往會經歷如下幾個階段

在這里插入圖片描述

所以要保證消息的可靠投遞，只需要保證這3個階段的可靠投遞即可

生產階段

這個階段的可靠投遞主要靠ConfirmListener(發布者確認)和ReturnListener(失敗通知)

前面已經介紹過了，一條消息在RabbitMQ中的流轉過程為

producer -> rabbitmq broker cluster -> exchange -> queue ->  consumer

ConfirmListener可以獲取消息是否從producer發送到broker

ReturnListener可以獲取從exchange路由不到queue的消息

我用Spring Boot Starter 的api來演示一下效果

application.yaml

spring:   rabbitmq:     host: myhost     port: 5672     username: guest     password: guest     virtual-host: /     listener:       simple:         acknowledge-mode: manual # 手動ack，默認為auto  log:   exchange: log.exchange   info:     queue: info.log.queue     binding-key: info.log.key

發布者確認回調

@Component public class ConfirmCallback implements RabbitTemplate.ConfirmCallback {      @Autowired     private MessageSender messageSender;      @Override     public void confirm(CorrelationData correlationData, boolean ack, String cause) {         String msgId = correlationData.getId();         String msg = messageSender.dequeueUnAckMsg(msgId);         if (ack) {             System.out.println(String.format("消息 {%s} 成功發送給mq", msg));         } else {             // 可以加一些重試的邏輯             System.out.println(String.format("消息 {%s} 發送mq失敗", msg));         }     } }

失敗通知回調

@Component public class ReturnCallback implements RabbitTemplate.ReturnCallback {      @Override     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {         String msg = new String(message.getBody());         System.out.println(String.format("消息 {%s} 不能被正確路由，routingKey為 {%s}", msg, routingKey));     } }

@Configuration public class RabbitMqConfig {      @Bean     public ConnectionFactory connectionFactory(             @Value("${spring.rabbitmq.host}") String host,             @Value("${spring.rabbitmq.port}") int port,             @Value("${spring.rabbitmq.username}") String username,             @Value("${spring.rabbitmq.password}") String password,             @Value("${spring.rabbitmq.virtual-host}") String vhost) {         CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host);         connectionFactory.setPort(port);         connectionFactory.setUsername(username);         connectionFactory.setPassword(password);         connectionFactory.setVirtualHost(vhost);         connectionFactory.setPublisherConfirms(true);         connectionFactory.setPublisherReturns(true);         return connectionFactory;     }      @Bean     public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,                                          ReturnCallback returnCallback, ConfirmCallback confirmCallback) {         RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);         rabbitTemplate.setReturnCallback(returnCallback);         rabbitTemplate.setConfirmCallback(confirmCallback);         // 要想使 returnCallback 生效，必須設置為true         rabbitTemplate.setMandatory(true);         return rabbitTemplate;     } }

這里我對RabbitTemplate做了一下包裝，主要就是發送的時候增加消息id，并且保存消息id和消息的對應關系，因為RabbitTemplate.ConfirmCallback只能拿到消息id，并不能拿到消息內容，所以需要我們自己保存這種映射關系。在一些可靠性要求比較高的系統中，你可以將這種映射關系存到數據庫中，成功發送刪除映射關系，失敗則一直發送

@Component public class MessageSender {      @Autowired     private RabbitTemplate rabbitTemplate;      public final Map<String, String> unAckMsgQueue = new ConcurrentHashMap<>();      public void convertAndSend(String exchange, String routingKey, String message) {         String msgId = UUID.randomUUID().toString();         CorrelationData correlationData = new CorrelationData();         correlationData.setId(msgId);         rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);         unAckMsgQueue.put(msgId, message);     }      public String dequeueUnAckMsg(String msgId) {         return unAckMsgQueue.remove(msgId);     }  }

測試代碼為

@RunWith(SpringRunner.class) @SpringBootTest public class MsgProducerTest {      @Autowired     private MessageSender messageSender;     @Value("${log.exchange}")     private String exchange;     @Value("${log.info.binding-key}")     private String routingKey;      /**      * 測試失敗通知      */     @SneakyThrows     @Test     public void sendErrorMsg() {         for (int i = 0; i < 3; i++) {             String message = "this is error message " + i;             messageSender.convertAndSend(exchange, "test", message);         }         System.in.read();     }      /**      * 測試發布者確認      */     @SneakyThrows     @Test     public void sendInfoMsg() {         for (int i = 0; i < 3; i++) {             String message = "this is info message " + i;             messageSender.convertAndSend(exchange, routingKey, message);         }         System.in.read();     } }

先來測試失敗者通知

輸出為

消息 {this is error message 0} 不能被正確路由，routingKey為 {test} 消息 {this is error message 0} 成功發送給mq 消息 {this is error message 2} 不能被正確路由，routingKey為 {test} 消息 {this is error message 2} 成功發送給mq 消息 {this is error message 1} 不能被正確路由，routingKey為 {test} 消息 {this is error message 1} 成功發送給mq

消息都成功發送到broker，但是并沒有被路由到queue中

再來測試發布者確認

輸出為

消息 {this is info message 0} 成功發送給mq infoLogQueue 收到的消息為: {this is info message 0} infoLogQueue 收到的消息為: {this is info message 1} 消息 {this is info message 1} 成功發送給mq infoLogQueue 收到的消息為: {this is info message 2} 消息 {this is info message 2} 成功發送給mq

看完上述內容，你們對RabbitMQ中怎么保證消息的可靠投遞有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業資訊頻道，感謝大家的支持。