RabbitMQ死信隊列&延遲交換機

1.什么是死信

死信&死信隊列

死信隊列的應用：

• 基于死信隊列在隊列消息已滿的情況下，消息也不會丟失
• 實現(xiàn)延遲消費的效果。比如：下訂單時，有15分鐘的付款時間

2. 實現(xiàn)死信隊列

2.1 準備Exchange&Queue

package com.llp.rabbitmq.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 死信隊列配置*/
@Configuration
public class DeadLetterConfig {public static final String NORMAL_EXCHANGE = "normal-exchange";public static final String NORMAL_QUEUE = "normal-queue";public static final String NORMAL_ROUTING_KEY = "normal.#";public static final String DEAD_EXCHANGE = "dead-exchange";public static final String DEAD_QUEUE = "dead-queue";public static final String DEAD_ROUTING_KEY = "dead.#";@Beanpublic Exchange normalExchange(){return ExchangeBuilder.topicExchange(NORMAL_EXCHANGE).build();}@Beanpublic Queue normalQueue(){//普通隊列，綁定死信隊列return QueueBuilder.durable(NORMAL_QUEUE).deadLetterExchange(DEAD_EXCHANGE).deadLetterRoutingKey("dead.abc").build();}@Beanpublic Binding normalBinding(Queue normalQueue,Exchange normalExchange){return BindingBuilder.bind(normalQueue).to(normalExchange).with(NORMAL_ROUTING_KEY).noargs();}@Beanpublic Exchange deadExchange(){return ExchangeBuilder.topicExchange(DEAD_EXCHANGE).build();}@Beanpublic Queue deadQueue(){return QueueBuilder.durable(DEAD_QUEUE).build();}@Beanpublic Binding deadBinding(Queue deadQueue,Exchange deadExchange){return BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY).noargs();}
}

2.2 實現(xiàn)效果

• 基于消費者進行reject或者nack實現(xiàn)死信效果

  package com.llp.rabbitmq.topic;import com.llp.rabbitmq.config.DeadLetterConfig;
  import com.rabbitmq.client.Channel;
  import org.springframework.amqp.core.Message;
  import org.springframework.amqp.rabbit.annotation.RabbitListener;
  import org.springframework.stereotype.Component;import java.io.IOException;@Component
  public class DeadListener {@RabbitListener(queues = DeadLetterConfig.NORMAL_QUEUE)public void consume(String msg, Channel channel, Message message) throws IOException {System.out.println("接收到normal隊列的消息：" + msg);//設置消息決絕消費，不需要重新放入到隊列中channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);//或者//channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);}
  }
  

• 消息的生存時間

  • 給消息設置生存時間

    @Test
    public void publishExpire(){String msg = "dead letter expire";rabbitTemplate.convertAndSend(DeadLetterConfig.NORMAL_EXCHANGE, "normal.abc", msg, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setExpiration("5000");return message;}});
    }
    

  • 給隊列設置消息的生存時間

    @Bean
    public Queue normalQueue(){return QueueBuilder.durable(NORMAL_QUEUE).deadLetterExchange(DEAD_EXCHANGE).deadLetterRoutingKey("dead.abc").ttl(10000).build();
    }
    

• 設置Queue中的消息最大長度

  @Bean
  public Queue normalQueue(){return QueueBuilder.durable(NORMAL_QUEUE).deadLetterExchange(DEAD_EXCHANGE).deadLetterRoutingKey("dead.abc").maxLength(1).build();
  }
  

  只要Queue中已經(jīng)有一個消息，如果再次發(fā)送一個消息，這個消息會變?yōu)樗佬?#xff01;

3.延遲交換機

下載地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/3.8.9

死信隊列實現(xiàn)延遲消費時，如果延遲時間比較復雜，比較多，直接使用死信隊列時，需要創(chuàng)建大量的隊列還對應不同的時間，可以采用延遲交換機來解決這個問題。

將下載的文件上傳到linux服務器并使用如下指令，將文件方到rabbitmq容器的plugins目錄下

docker cp rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez 9da57c5038ba:/opt/rabbitmq/plugins

在rabbitmq容器的/opt/rabbitmq/sbin目錄下執(zhí)行

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

重啟容器生效

docker restart 9da57c5038ba

可以看到添加插件后多了一個延遲交換機的選項

• 構建延遲交換機

  package com.llp.rabbitmq.config;import org.springframework.amqp.core.*;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Configuration;import java.util.HashMap;
  import java.util.Map;/*** 延遲隊列*/
  @Configuration
  public class DelayedConfig {public static final String DELAYED_EXCHANGE = "delayed-exchange";public static final String DELAYED_QUEUE = "delayed-queue";public static final String DELAYED_ROUTING_KEY = "delayed.#";@Beanpublic Exchange delayedExchange(){Map<String, Object> arguments = new HashMap<>();arguments.put("x-delayed-type","topic");Exchange exchange = new CustomExchange(DELAYED_EXCHANGE,"x-delayed-message",true,false,arguments);return exchange;}@Beanpublic Queue delayedQueue(){return QueueBuilder.durable(DELAYED_QUEUE).build();}@Beanpublic Binding delayedBinding(Queue delayedQueue,Exchange delayedExchange){return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();}
  }

• 發(fā)送消息

  package com.llp.rabbitmq;import com.llp.rabbitmq.config.DelayedConfig;
  import org.junit.jupiter.api.Test;
  import org.springframework.amqp.AmqpException;
  import org.springframework.amqp.core.Message;
  import org.springframework.amqp.core.MessagePostProcessor;
  import org.springframework.amqp.rabbit.core.RabbitTemplate;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.boot.test.context.SpringBootTest;@SpringBootTest
  public class DelayedPublisherTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void publish(){rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "delayed.abc", "xxxx", new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {//設置消息指定多少時間被消費，單位毫秒message.getMessageProperties().setDelay(30000);return message;}});}
  }
  

**延遲交換機存在的問題：**在延遲推送消息的過程中rabbitmq重啟了、或者說服務器宕機了就會導致消息丟失