详解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:study@47.98.109.138: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) 只适⽤特定版本