Spring Boot系列——死信隊列

在說死信隊列之前，我們先介紹下為什么需要用死信隊列。

如果想直接了解死信對接，直接跳入下文的"死信隊列"部分即可。

ack機制和requeue-rejected屬性

我們還是基於上篇《Spring Boot系列——7步集成RabbitMQ》的demo代碼來說。

在項目springboot-demo我們看到application.yaml文件部分配置內容如下

...

listener:
    type: simple
    simple:
      acknowledge-mode: auto
      concurrency: 5
      default-requeue-rejected: true
      max-concurrency: 100
...

其中

acknowledge-mode

該配置項是用來表示消息確認方式，其有三種配置方式，分別是none、manual和auto。

none意味着沒有任何的應答會被發送。

manual意味着監聽者必須通過調用Channel.basicAck()來告知所有的消息。

auto意味着容器會自動應答，除非MessageListener拋出異常，這是默認配置方式。

default-requeue-rejected

該配置項是決定由於監聽器拋出異常而拒絕的消息是否被重新放回隊列。默認值為true。

我一開始對於這個屬性有個誤解，我以為rejected是表示拒絕，所以將requeue-rejected連起來是拒絕重新放回隊列，后來查了資料明白這個屬性的功能才想起來rejected是個形容詞，其表示的應該是被拒絕的消息

所以如果該屬性配置為true表示會重新放回隊列，如果配置為false表示不會放回隊列。

下面我們看看acknowledge-mode參數和default-requeue-rejected參數使用不同的組合方式，RabbitMQ是如何處理消息的。

代碼依然使用springboot-demo中的RabbitApplicationTests發送消息，使用Receiver類監聽demo-queue隊列的消息。

對於Receiver類添加了一行代碼，該代碼模擬拋出異常

@Component
public class Receiver {

    @RabbitListener(queues = "demo_queue")
    public void created(String message) {
        System.out.println("orignal message: " + message);
        int i = 1/0;
    }
}

acknowledge-mode=none, default-requeue-rejected=false

該配置不會確認消息是否正常消費，所以在控制台沒有拋出任何異常。通過在RabbitMQ管理頁面也沒有看到重新放回隊列的消息

acknowledge-mode=none, default-requeue-rejected=true

同樣該配置不會確認消息是否正常消費，所以在控制台沒有拋出任何異常。而且即使default-requeue-rejected配置為true因為沒有確認所以也沒有看到重新放回隊列的消息

acknowledge-mode=manual, default-requeue-rejected=false

該配置需要手動確認消息是否正常消費，但是代碼中並沒有手動確認，個人理解是因為沒有收到ack，所以消息又回到了隊列中。

acknowledge-mode=manual, default-requeue-rejected=true

該配置需要手動確認消息是否正常消費，但是代碼中並沒有手動確認，所以消息被重新放入到隊列中了，並且在控制台發現還拋出了異常（這塊不是很清楚，default-requeue-rejected設置true和false帶來的不同效果，有了解的麻煩下方留言指教）。

acknowledge-mode=auto, default-requeue-rejected=false

該配置采用自動確認，從結果來看，是自動確認了。

從控制台打印的結果可以看出Receiver方法執行了3次，分別是前面兩條放回隊列的消息以及這次發送的消息，所以3條消息都消費了。

同時因為default-requeue-rejected設置為false，所以即使消費拋出異常，也沒有將消息放回隊列。

acknowledge-mode=auto, default-requeue-rejected=true

該配置同樣采用自動確認，從結果看出，沒有拋出異常（這塊也不是很理解），且因為default-requeue-rejected設置為true，所以消息重新回到隊列。

綜上羅列這么多情況只為說明有些情況下，如果消息消費出錯，因為配置問題導致消息丟失了。這在很多情況下是要命的，比如用戶支付的訂單號，如果因為拋異常等原因直接丟失是很要命的。

所以，我們需要有一個確保機制，能夠保證即使失敗的消息也能保存下來，這時候死信隊列就排上用場了。

死信隊列

死信隊列的整個設計思路是這樣的

生產者 --> 消息 --> 交換機 --> 隊列 --> 變成死信 --> DLX交換機 -->隊列 --> 消費者

下面我們通過網上的一個簡單的死信隊列的實現看看如何使用死信隊列。

@Bean("deadLetterExchange")
    public Exchange deadLetterExchange() {
        return ExchangeBuilder.directExchange("DL_EXCHANGE").durable(true).build();
    }

    @Bean("deadLetterQueue")
    public Queue deadLetterQueue() {
        Map<String, Object> args = new HashMap<>(2);
//       x-dead-letter-exchange    聲明  死信交換機
        args.put("x-dead-letter-exchange", "DL_EXCHANGE");
//       x-dead-letter-routing-key    聲明 死信路由鍵
        args.put("x-dead-letter-routing-key", "KEY_R");
        return QueueBuilder.durable("DL_QUEUE").withArguments(args).build();
    }

    @Bean("redirectQueue")
    public Queue redirectQueue() {
        return QueueBuilder.durable("REDIRECT_QUEUE").build();
    }

    /**
     * 死信路由通過 DL_KEY 綁定鍵綁定到死信隊列上.
     *
     * @return the binding
     */
    @Bean
    public Binding deadLetterBinding() {
        return new Binding("DL_QUEUE", Binding.DestinationType.QUEUE, "DL_EXCHANGE", "DL_KEY", null);

    }

    /**
     * 死信路由通過 KEY_R 綁定鍵綁定到死信隊列上.
     *
     * @return the binding
     */
    @Bean
    public Binding redirectBinding() {
        return new Binding("REDIRECT_QUEUE", Binding.DestinationType.QUEUE, "DL_EXCHANGE", "KEY_R", null);
    }

注意

• 聲明了一個direct模式的exchange。

• 聲明了一個死信隊列deadLetterQueue，該隊列配置了一些屬性x-dead-letter-exchange表明死信交換機，x-dead-letter-routing-key表明死信路由鍵，因為是direct模式，所以需要設置這個路由鍵。

• 聲明了一個替補隊列redirectQueue，變成死信的消息最終就是存放在這個隊列的。

• 聲明綁定關系，分別是死信隊列以及替補隊列和交換機的綁定。

那么如何模擬生成一個死信消息呢，可以在發送到DL_QUEUE的消息在10秒后失效，然后轉發到替補隊列中，代碼實現如下

public void sendMsg(String content) {
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        MessagePostProcessor messagePostProcessor = message -> {
            MessageProperties messageProperties = message.getMessageProperties();
//            設置編碼
            messageProperties.setContentEncoding("utf-8");
//            設置過期時間10*1000毫秒
            messageProperties.setExpiration("5000");
            return message;
        };
        rabbitTemplate.convertAndSend("DL_EXCHANGE", "DL_KEY", content, messagePostProcessor);
    }

執行結果如下

消息首先進入DL_QUEUE，5秒后失效，被轉發到REDIRECT_QUEUE中。

如果您覺得閱讀本文對您有幫助，請點一下“推薦”按鈕，您的“推薦”將是我最大的寫作動力！如果您想持續關注我的文章，請掃描二維碼，關注JackieZheng的微信公眾號，我會將我的文章推送給您，並和您一起分享我日常閱讀過的優質文章。