環(huán)境：SpringBoot2.7.9

10年的揭西網(wǎng)站建設(shè)經(jīng)驗，針對設(shè)計、前端、開發(fā)、售后、文案、推廣等六對一服務(wù)，響應(yīng)快，48小時及時工作處理。成都全網(wǎng)營銷推廣的優(yōu)勢是能夠根據(jù)用戶設(shè)備顯示端的尺寸不同，自動調(diào)整揭西建站的顯示方式，使網(wǎng)站能夠適用不同顯示終端，在瀏覽器中調(diào)整網(wǎng)站的寬度，無論在任何一種瀏覽器上瀏覽網(wǎng)站，都能展現(xiàn)優(yōu)雅布局與設(shè)計，從而大程度地提升瀏覽體驗。創(chuàng)新互聯(lián)公司從事“揭西網(wǎng)站設(shè)計”,“揭西網(wǎng)站推廣”以來，每個客戶項目都認真落實執(zhí)行。

消息丟失場景

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

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

1. 通過事務(wù)（不推薦）
2. 確認機制（推薦）

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

• 引入依賴

  org.springframework.boot
  spring-boot-starter-amqp

• 聲明交換機及隊列

@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配置對應(yīng)的Callback，Publisher確認回調(diào)，Publisher返回回調(diào)。

1. 確認回調(diào)
   當消息發(fā)送到了交換機則ack=true，當消息無法發(fā)送到交換機則ack=false。
2. 返回回調(diào)
   當消息能夠發(fā)送到交換機，但是不能路由到隊列則會調(diào)用該return回調(diào)。

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

1. 自定義RabbitTemplate。
2. 通過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ā)送到交換機") ;
        } 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()) ;
      }
    });
  }

}

使用錯誤的交換機和錯誤的路由key分別測試即可以看到上面的輸出信息了。

MQ Server丟消息

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

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

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

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

消費者丟消息

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

默認是自動應(yīng)答，當消息監(jiān)聽方法中沒有異常時則正常應(yīng)答，當發(fā)生異常時，在默認情況下會重新入隊列（這樣就會出現(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è)置為手動應(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ù)代碼
    // 當消費者把消息消費成功，再手動應(yīng)答RabbitMQ
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  } catch (Exception e) {
    // 如果發(fā)生了異常，我們一般的處理是直接扔掉死信隊列，一般這里出現(xiàn)錯誤都是消息有問題
    // 如果消息出現(xiàn)問題，你重試再入隊列是無意義的
  }
}

消息重試

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

修改配置：

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次間的重試最大間隔時間
          maxInterval: 20000
          maxAttempts: 4 #重試4次，1s, 3s, 9s
          stateless: true #如果消息處理中存在事務(wù)則需要將其設(shè)置為false

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

1. ImmediateRequeueMessageRecoverer
2. RejectAndDontRequeueRecoverer
3. RepublishMessageRecoverer

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

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

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

網(wǎng)站題目：SpringBoot整合RabbitMQ保證消息的可靠的投遞及消費
文章源于：http://uogjgqi.cn/article/dpihshc.html