前言

在說(shuō)死信隊(duì)列之前，我們先介紹下為什么需要用死信隊(duì)列。

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

ack機(jī)制和requeue-rejected屬性

我們還是基于上篇《Spring Boot系列——7步集成RabbitMQ》的demo代碼來(lái)說(shuō)。

在項(xiàng)目springboot-demo我們看到application.yaml文件部分配置內(nèi)容如下

...

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

其中

acknowledge-mode

該配置項(xiàng)是用來(lái)表示消息確認(rèn)方式，其有三種配置方式，分別是none、manual和auto。

none意味著沒(méi)有任何的應(yīng)答會(huì)被發(fā)送。

manual意味著監(jiān)聽(tīng)者必須通過(guò)調(diào)用Channel.basicAck()來(lái)告知所有的消息。

auto意味著容器會(huì)自動(dòng)應(yīng)答，除非MessageListener拋出異常，這是默認(rèn)配置方式。

default-requeue-rejected

該配置項(xiàng)是決定由于監(jiān)聽(tīng)器拋出異常而拒絕的消息是否被重新放回隊(duì)列。默認(rèn)值為true。

我一開(kāi)始對(duì)于這個(gè)屬性有個(gè)誤解，我以為rejected是表示拒絕，所以將requeue-rejected連起來(lái)是拒絕重新放回隊(duì)列，后來(lái)查了資料明白這個(gè)屬性的功能才想起來(lái)rejected是個(gè)形容詞，其表示的應(yīng)該是被拒絕的消息

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

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

代碼依然使用springboot-demo中的RabbitApplicationTests發(fā)送消息，使用Receiver類(lèi)監(jiān)聽(tīng)demo-queue隊(duì)列的消息。

對(duì)于Receiver類(lèi)添加了一行代碼，該代碼模擬拋出異常

@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

該配置不會(huì)確認(rèn)消息是否正常消費(fèi)，所以在控制臺(tái)沒(méi)有拋出任何異常。通過(guò)在RabbitMQ管理頁(yè)面也沒(méi)有看到重新放回隊(duì)列的消息

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

同樣該配置不會(huì)確認(rèn)消息是否正常消費(fèi)，所以在控制臺(tái)沒(méi)有拋出任何異常。而且即使default-requeue-rejected配置為true因?yàn)闆](méi)有確認(rèn)所以也沒(méi)有看到重新放回隊(duì)列的消息

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

該配置需要手動(dòng)確認(rèn)消息是否正常消費(fèi)，但是代碼中并沒(méi)有手動(dòng)確認(rèn)，個(gè)人理解是因?yàn)闆](méi)有收到ack，所以消息又回到了隊(duì)列中。

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

該配置需要手動(dòng)確認(rèn)消息是否正常消費(fèi)，但是代碼中并沒(méi)有手動(dòng)確認(rèn)，所以消息被重新放入到隊(duì)列中了，并且在控制臺(tái)發(fā)現(xiàn)還拋出了異常（這塊不是很清楚，default-requeue-rejected設(shè)置true和false帶來(lái)的不同效果，有了解的麻煩下方留言指教）。

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

該配置采用自動(dòng)確認(rèn)，從結(jié)果來(lái)看，是自動(dòng)確認(rèn)了。

從控制臺(tái)打印的結(jié)果可以看出Receiver方法執(zhí)行了3次，分別是前面兩條放回隊(duì)列的消息以及這次發(fā)送的消息，所以3條消息都消費(fèi)了。

同時(shí)因?yàn)閐efault-requeue-rejected設(shè)置為false，所以即使消費(fèi)拋出異常，也沒(méi)有將消息放回隊(duì)列。

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

該配置同樣采用自動(dòng)確認(rèn)，從結(jié)果看出，沒(méi)有拋出異常（這塊也不是很理解），且因?yàn)閐efault-requeue-rejected設(shè)置為true，所以消息重新回到隊(duì)列。

綜上羅列這么多情況只為說(shuō)明有些情況下，如果消息消費(fèi)出錯(cuò)，因?yàn)榕渲脝?wèn)題導(dǎo)致消息丟失了。這在很多情況下是要命的，比如用戶(hù)支付的訂單號(hào)，如果因?yàn)閽伄惓５仍蛑苯觼G失是很要命的。

所以，我們需要有一個(gè)確保機(jī)制，能夠保證即使失敗的消息也能保存下來(lái)，這時(shí)候死信隊(duì)列就排上用場(chǎng)了。

死信隊(duì)列

死信隊(duì)列的整個(gè)設(shè)計(jì)思路是這樣的

生產(chǎn)者 --> 消息 --> 交換機(jī) --> 隊(duì)列 --> 變成死信 --> DLX交換機(jī) -->隊(duì)列 --> 消費(fèi)者

下面我們通過(guò)網(wǎng)上的一個(gè)簡(jiǎn)單的死信隊(duì)列的實(shí)現(xiàn)看看如何使用死信隊(duì)列。

@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 聲明 死信交換機(jī)
  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();
 }

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

 }

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

注意

聲明了一個(gè)direct模式的exchange。

聲明了一個(gè)死信隊(duì)列deadLetterQueue，該隊(duì)列配置了一些屬性x-dead-letter-exchange表明死信交換機(jī)，x-dead-letter-routing-key表明死信路由鍵，因?yàn)槭莇irect模式，所以需要設(shè)置這個(gè)路由鍵。

聲明了一個(gè)替補(bǔ)隊(duì)列redirectQueue，變成死信的消息最終就是存放在這個(gè)隊(duì)列的。

聲明綁定關(guān)系，分別是死信隊(duì)列以及替補(bǔ)隊(duì)列和交換機(jī)的綁定。

那么如何模擬生成一個(gè)死信消息呢，可以在發(fā)送到DL_QUEUE的消息在10秒后失效，然后轉(zhuǎn)發(fā)到替補(bǔ)隊(duì)列中，代碼實(shí)現(xiàn)如下

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

執(zhí)行結(jié)果如下

消息首先進(jìn)入DL_QUEUE，5秒后失效，被轉(zhuǎn)發(fā)到REDIRECT_QUEUE中。

總結(jié)

以上就是這篇文章的全部?jī)?nèi)容了，希望本文的內(nèi)容對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，如果有疑問(wèn)大家可以留言交流，謝謝大家對(duì)腳本之家的支持。

最新評(píng)論