详解RabbitMQ高级特性之延迟插件的安装和使用
目录
延迟队列插件
RabbitMQ官⽅也提供了⼀个延迟的插件来实现延迟的功能。
延迟队列插件的下载
插件下载链接---》链接
根据⾃⼰的RabbitMQ版本选择相应版本的延迟插件, 下载后上传到服务器。
查看RabbitMQ版本的命令
rabbitmqctl status | grep "RabbitMQ"
/usr/lib/rabbitmq/plugins 是⼀个附加⽬录, RabbitMQ包本⾝不会在此安装任何内容, 如果
没有这个路径, 可以⾃⼰进⾏创建。
延迟队列插件的安装
把下载好的 .ez文件上传到 /usr/lib/rabbitmq/plugins 这个路径下。
查看rabbitmq插件的命令
rabbitmq-plugins list
延迟队列插件的启用
启用延迟队列插件的命令
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
添加配置
spring: application: name: rabbit-extensions-demo rabbitmq: addresses: amqp://study:[email protected]:5672/extension
常量类
public class Constants { //延迟队列 public static final String DELAY_QUEUE = "delay.queue"; public static final String DELAY_EXCHANGE = "delay.exchange"; }声明队列和交换机并绑定二者关系
import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import rabbitextensionsdemo.constant.Constants; @Configuration public class DelayConfig { @Bean("delayQueue") public Queue delayQueue(){ return QueueBuilder.durable(Constants.DELAY_QUEUE).build(); } @Bean("delayExchange") public Exchange delayExchange(){ return ExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE).delayed().build(); } @Bean("delayBinding") public Binding delayBinding(@Qualifier("delayQueue") Queue queue, @Qualifier("delayExchange") Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with("delay").noargs(); } }编写生产消息代码
@RequestMapping("/delay2") public String delay2() { System.out.println("delay2..."); rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, "delay", "delay test 30s...", message -> { message.getMessageProperties().setDelayLong(30000L); //单位: 毫秒, 过期时间为30s return message; }); rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, "delay", "delay test 10s...", message -> { message.getMessageProperties().setDelayLong(10000L); //单位: 毫秒, 延迟时间为10s return message; }); System.out.printf("%tc 消息发送成功 \n", new Date()); return "消息发送成功"; }编写消费消息代码
import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; import rabbitextensionsdemo.constant.Constants; import java.util.Date; @Component public class DelayListener { @RabbitListener(queues = Constants.DELAY_QUEUE) public void delayHandMessage(Message message, Channel channel) throws Exception { //消费者逻辑 System.out.printf("[delay.queue] %tc 接收到消息: %s \n", new Date(), new String(message.getBody(),"UTF-8")); } }观察效果
此时我们可以看到，哪怕把TTL为30秒的消息先于TTL为10秒的消息进行发送，也不会影响根据延迟时间进行消息的发送。
面试题
1.介绍下RabbitMQ的延迟队列
延迟队列是⼀个特殊的队列, 消息发送之后, 并不⽴即给消费者, ⽽是等待特定的时间, 才发送给消费者.
延迟队列的应⽤场景有很多, ⽐如：
1. 订单在⼗分钟内未⽀付⾃动取消
2. ⽤⼾注册成功后, 3天后发调查问卷
3. ⽤⼾发起退款, 24⼩时后商家未处理, 则默认同意, ⾃动退款
但RabbitMQ本⾝并没直接实现延迟队列, 通常有两种⽅法：
1. TTL+死信队列组合的⽅式
2. 使⽤官⽅提供的延迟插件实现延迟功能
2.上面两种实现方法二者对比
1. 基于死信实现的延迟队列
a. 优点: 1) 灵活不需要额外的插件⽀持
b. 缺点: 1) 存在消息顺序问题 2) 需要额外的逻辑来处理死信队列的消息, 增加了系统的复杂性
2. 基于插件实现的延迟队列
a. 优点: 1) 通过插件可以直接创建延迟队列, 简化延迟消息的实现. 2) 避免了DLX的时序问题
b. 缺点: 1) 需要依赖特定的插件, 有运维⼯作 2) 只适⽤特定版本
