RabbitMQ - 延迟消息 - 死信交换机

avatar
作者
筋斗云
阅读量:2

目录

1、怎么理解延迟消息?

2、如何实现延迟消息?

2.1、方案一:死信交换机

2.1.1、什么是死信:

 2.1.2、什么是死信交换机?

2.2、方案二:延迟消息插件

2.2.1、插件安装:

2.2.2、代码实现


1、怎么理解延迟消息?

  • 延迟消息:生产者发送消息时指定一个时间,消费者不会立刻收到消息,而是在指定时间之后才收到消息
  • 延时任务:设置在一定时间之后才执行的任务

2、如何实现延迟消息?

2.1、方案一:死信交换机

2.1.1、什么是死信:

        当一个队列中的消息满足下列情况之一时,就会成为死信(Dead Letter):        

  • 消费者使用basic.reject或basic.nack声明消费失败,并且消息的Requeue参数设置为false
  • 消息是一个过期消息(达到了队列或消息本身设置的过期时间),超时无人消费
  • 要投递的队列消息堆积满了(可以给队列设置一个消息上限),最早的消息可能成为死信       

 2.1.2、什么是死信交换机?

        如果队列通过dead-letter-exchange属性指定了一个交换机,那么该队列中的死信就会投递到这个交换机中。这个交换机称为死信交换机(Dead Letter Exchange,简称DLX)

如图理解:

        图解: 我们发一个消息,过期时间为30s,由于simple.queue没有消费者,时间到了之后,simple.queue就会把死信投递给dlx.direct交换机~

(官方推出的死信交换机并不是用来实现延迟消息的,而是用做一些兜底方案,方便人工介入的~)

2.2、方案二:延迟消息插件

        RabbitMQ的官方推出一个插件,原生支持延迟消息功能。该插件的原理是设计了一种支持延迟消息功能的交换机,当消息投递到交换机后可以暂存一定时间,到期后再投递到队列中~

2.2.1、插件安装:

官网:Community Plugins | RabbitMQ

提示:RabbitMQ是什么版本的,下载的插件就得是什么版本的,得对应上

等待下载,下载好了之后,把这个插件传输到服务器上,放在哪儿?

把他放在官网提示的目录下:

我把目录已经挂载了,就使用本地目录啦~

放到指定目录后执行一条命令:

docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

执行完毕后,验证是否安装成功:

2.2.2、代码实现

声明延迟交换机的两个方案:

方案一:注解:

@RabbitListener(bindings = @QueueBinding(         value = @Queue(name = "delay.queue", durable = "true"),         exchange = @Exchange(name = "delay.direct", delayed = "true"),         key = "delay" )) public void listenDelayMessage(String msg){     log.info("接收到delay.queue的延迟消息:{}", msg); }

方案二:@Bean:

package com.itheima.consumer.config;  import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;  @Slf4j @Configuration public class DelayExchangeConfig {      @Bean     public DirectExchange delayExchange(){         return ExchangeBuilder                 .directExchange("delay.direct") // 指定交换机类型和名称                 .delayed() // 设置delay的属性为true                 .durable(true) // 持久化                 .build();     }      @Bean     public Queue delayedQueue(){         return new Queue("delay.queue");     }          @Bean     public Binding delayQueueBinding(){         return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");     } } 

发送延迟消息
发送消息时,必须通过x-delay属性设定延迟时间:

@Test void testPublisherDelayMessage() {     // 1.创建消息     String message = "hello, delayed message";     // 2.发送消息,利用消息后置处理器添加消息头     rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {         @Override         public Message postProcessMessage(Message message) throws AmqpException {             // 添加延迟消息属性             message.getMessageProperties().setDelay(5000);             return message;         }     }); }

广告一刻

为您即时展示最新活动产品广告消息,让您随时掌握产品活动新动态!