SpringBoot整合RabbitMQ保證消息的可靠的投遞及消費(fèi)

作者：Spring全家桶實(shí)戰(zhàn)案例 2023-03-06 08:16:04

開發(fā) 架構(gòu)

Spring.rabbitmq.publisher-confirm-type及spring.rabbitmq.publisher-returns 的配置值。

環(huán)境：SpringBoot2.7.9

消息丟失場(chǎng)景

生產(chǎn)者丟失消息
生產(chǎn)者發(fā)出的數(shù)據(jù)由于網(wǎng)絡(luò)原因沒有到底MQ Server丟失
MQ Server丟消息
由于消息隊(duì)列沒有持久化或者是消息沒有持久化，在Server重啟后消息丟失
消費(fèi)者丟消息
接收到消息后，業(yè)務(wù)還沒有處理完成，服務(wù)宕機(jī)（當(dāng)你是自動(dòng)ACK）。

生產(chǎn)者丟失解決方案

通過事務(wù)（不推薦）
確認(rèn)機(jī)制（推薦）

這里只講如何通過確認(rèn)機(jī)制保證生產(chǎn)者不丟失消息

引入依賴

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

聲明交換機(jī)及隊(duì)列

@Bean
public TopicExchange topicExchange() {
  return new TopicExchange("akf.exchange", true, false) ;
}
@Bean
public Queue queue() {
  return new Queue("akf.queue", true, false, false) ;
}
@Bean
public Binding binding() {
  return BindingBuilder.bind(queue()).to(topicExchange()).with("akf.#") ;
}

RabbitMQ配置

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtualHost: test
    publisherConfirmType: correlated  
    publisherReturns: true
    template:
      mandatory: true

注意：spring.rabbitmq.publisher-confirm-type及spring.rabbitmq.publisher-returns 的配置值。

接下來是為RabbitTemplate配置對(duì)應(yīng)的Callback，Publisher確認(rèn)回調(diào)，Publisher返回回調(diào)。

確認(rèn)回調(diào)
當(dāng)消息發(fā)送到了交換機(jī)則ack=true，當(dāng)消息無法發(fā)送到交換機(jī)則ack=false。
返回回調(diào)
當(dāng)消息能夠發(fā)送到交換機(jī)，但是不能路由到隊(duì)列則會(huì)調(diào)用該return回調(diào)。

RabbitTemplate是單例的可以通過兩種方式配置對(duì)應(yīng)的回調(diào)。

自定義RabbitTemplate。
通過AWare接口獲取RabbitTemplate配置。

這里只講通過AWare接口配置回調(diào)。

配置Callback

@Component
public class ConfigRabbitTemplate implements ApplicationContextAware {

  @Override
  public void setApplicationContext(ApplicationContext context) throws BeansException {
    RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class) ;
    rabbitTemplate.setConfirmCallback(new ConfirmCallback() {
      @Override
      public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("correlation: " + correlationData) ;
        if (ack) {
          System.out.println("消息發(fā)送到交換機(jī)") ;
        } else {
          System.out.println("消息發(fā)送失敗 - " + ", cause" + cause) ;
        }
      }
    });
    rabbitTemplate.setReturnsCallback(new ReturnsCallback() {
      @Override
      public void returnedMessage(ReturnedMessage returned) {
        System.out.println(returned.getExchange() + ", " + returned.getRoutingKey() + ", " + returned.getReplyCode() + ", " + returned.getMessage().toString()) ;
      }
    });
  }

}

使用錯(cuò)誤的交換機(jī)和錯(cuò)誤的路由key分別測(cè)試即可以看到上面的輸出信息了。

MQ Server丟消息

在通過@Bean聲明交換機(jī)和隊(duì)列時(shí)設(shè)置持久性，在消息上設(shè)置持久化。

@Bean
public TopicExchange topicExchange() {
  // 這里的第二個(gè)參數(shù)就是設(shè)置是否持久化，如果設(shè)置為false，當(dāng)服務(wù)重啟交換機(jī)將丟失
  // 第三個(gè)參數(shù)是否自動(dòng)刪除，當(dāng)不再使用該交換機(jī)時(shí)會(huì)自動(dòng)刪除該交換機(jī)
  return new TopicExchange("akf.exchange", true, false) ;
}
@Bean
public Queue queue() {
  // 第二個(gè)參數(shù)true設(shè)置隊(duì)列是持久化的，當(dāng)服務(wù)重啟隊(duì)列不會(huì)丟失
  return new Queue("akf.queue", true, false, false) ;
}

設(shè)置消息持久化。

Message message = MessageBuilder.withBody("Hello".getBytes())
  // 設(shè)置消息投遞模式為持久化的（默認(rèn)不設(shè)置就是持久化的）
  .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
  .build() ;

消費(fèi)者丟消息

關(guān)閉自動(dòng)應(yīng)答機(jī)制。

默認(rèn)是自動(dòng)應(yīng)答，當(dāng)消息監(jiān)聽方法中沒有異常時(shí)則正常應(yīng)答，當(dāng)發(fā)生異常時(shí)，在默認(rèn)情況下會(huì)重新入隊(duì)列（這樣就會(huì)出現(xiàn)死循環(huán)）。

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtualHost: test
    publisherConfirmType: correlated
    publisherReturns: true
    listener:
      simple:
        acknowledgeMode: manual #設(shè)置為手動(dòng)應(yīng)答

消息監(jiān)聽。

@RabbitListener(queues = {"akf.queue"})
public void onMessage(Message message, Channel channel) throws Exception {
  try {
    System.out.println("接收到消息： " + new String(message.getBody()));
    // ... 這里處理我們的業(yè)務(wù)代碼
    // 當(dāng)消費(fèi)者把消息消費(fèi)成功，再手動(dòng)應(yīng)答RabbitMQ
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  } catch (Exception e) {
    // 如果發(fā)生了異常，我們一般的處理是直接扔掉死信隊(duì)列，一般這里出現(xiàn)錯(cuò)誤都是消息有問題
    // 如果消息出現(xiàn)問題，你重試再入隊(duì)列是無意義的
  }
}

消息重試

如果消息消費(fèi)時(shí)出現(xiàn)錯(cuò)誤，你又希望能夠通過重試來盡可能的處理掉該消息，Spring也提供了相應(yīng)的重試機(jī)制。

修改配置：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtualHost: test
    publisherConfirmType: correlated
    publisherReturns: true
    listener:
      simple:
        acknowledgeMode: auto
        concurrency: 1
        retry:
          # 開啟重試
          enabled: true
          # 延遲1s后開始重試
          initialInterval: 1000
          # 每次消息重試的間隔乘數(shù)
          multiplier: 3
          # 2次間的重試最大間隔時(shí)間
          maxInterval: 20000
          maxAttempts: 4 #重試4次，1s, 3s, 9s
          stateless: true #如果消息處理中存在事務(wù)則需要將其設(shè)置為false

如果只是做上面的配置，重試指定次數(shù)后消息將會(huì)被丟棄，這是默認(rèn)行為。Spring提供了 MessageRecoverer接口來決定消息如何處理。默認(rèn)Spring提供如下幾種實(shí)現(xiàn)：

ImmediateRequeueMessageRecoverer
RejectAndDontRequeueRecoverer
RepublishMessageRecoverer

我們只需要定義一個(gè)Bean為MessageRecoverer即可，這里我們就用Spring提供的RepublishMessageRecoverer重新發(fā)布消息。

@Bean
public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
  return new RepublishMessageRecoverer(rabbitTemplate, "error.exchange", "error") ;
}

這里將消息重新發(fā)布一個(gè)專門的隊(duì)列（重試指定次數(shù)后）。

責(zé)任編輯：姜華來源：今日頭條

Spring RabbitMQ