SpringBoot | 第三十八章：基于RabbitMQ实现消息延迟队列方案

前言

• 前言
• 一点知识 延时队列使用场景
• RabbitMQ实现延时队列 存活时间（Time-To-Live 简称 TTL）
• 死信交换（Dead Letter Exchanges 简称 DLX）

一点知识

延时队列使用场景

1. 订单超时关闭：在支付场景中，一般上订单在创建后30分钟或1小时内未支付的，会自动取消订单。
2. 短信或者邮件通知：在一些注册或者下单业务时，需要在1分钟或者特定时间后进行短信或者邮件发送相关资料的。本身此类业务于主业务是无关联性的，一般上的做法是进行异步发送。
3. 重试场景：比如消息通知，在第一次通知出现异常时，会在隔几分钟之后进行再次重试发送。

RabbitMQ实现延时队列

RabbitMQ

TTL(Time-To-Live，存活时间)

DLX(Dead-Letter-Exchange，死信队列交换机)

存活时间（Time-To-Live 简称 TTL）

RabbitMQ

死亡

Dead Letter

死信交换（Dead Letter Exchanges 简称 DLX）

TTL

Dead Letter

一个消息变成死信一般上是由于以下几种情况;

1. 消息被拒绝
2. 消息过期
3. 队列达到了最大长度。

所以，通过

TTL

DLX

SpringBoot集成RabbitMQ实现延时队列实战

SpringBoot

RabbitMQ

http

0.引入pom依赖

<!-- rabbit -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 简化http操作 -->
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-http</artifactId>
<version>4.5.16</version>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-json</artifactId>
<version>4.5.16</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>

1.编写

rabbitmq

RabbitConfig.java

/**
*
* @ClassName   类名：RabbitConfig
* @Description 功能说明：
* <p>
* TODO
*</p>
************************************************************************
* @date        创建日期：2019年7月17日
* @author      创建人：oKong
* @version     版本号：V1.0
*<p>
***************************修订记录*************************************
*
*   2019年7月17日   oKong   创建该类功能。
*
***********************************************************************
*</p>
*/
@Configuration
public class RabbitConfig {

@Autowired
ConnectionFactory connectionFactory;

/**
* 消费者线程数 设置大点 大概率是能通知到的
*/
@Value("${http.notify.concurrency:50}")
int concurrency;

/**
* 延迟队列的消费者线程数 可设置小点
*/
@Value("${http.notify.delay.concurrency:20}")
int delayConcurrency;

@Bean
public RabbitAdmin rabbitAdmin() {
return new RabbitAdmin(connectionFactory);
}

@Bean
public DirectExchange httpMessageNotifyDirectExchange(RabbitAdmin rabbitAdmin) {
//durable 是否持久化
//autoDelete 是否自动删除，即服务端或者客服端下线后 交换机自动删除
DirectExchange directExchange = new DirectExchange(ApplicationConstant.HTTP_MESSAGE_EXCHANGE,true,false);
directExchange.setAdminsThatShouldDeclare(rabbitAdmin);
return directExchange;
}

//设置消息队列
@Bean
public Queue httpMessageStartQueue(RabbitAdmin rabbitAdmin) {

/*
创建接收队列，4个参数
name - 队列名称
durable - false，不进行持有化
exclusive - true，独占性
autoDelete - true，自动删除*/
Queue queue = new Queue(ApplicationConstant.HTTP_MESSAGE_START_QUEUE_NAME, true, false, false);
queue.setAdminsThatShouldDeclare(rabbitAdmin);
return queue;
}

//队列绑定交换机
@Bean
public Binding bindingStartQuene(RabbitAdmin rabbitAdmin,DirectExchange httpMessageNotifyDirectExchange, Queue httpMessageStartQueue) {
Binding binding = BindingBuilder.bind(httpMessageStartQueue).to(httpMessageNotifyDirectExchange).with(ApplicationConstant.HTTP_MESSAGE_START_RK);
binding.setAdminsThatShouldDeclare(rabbitAdmin);
return binding;
}

@Bean
public Queue httpMessageOneQueue(RabbitAdmin rabbitAdmin) {
Queue queue = new Queue(ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME, true, false, false);
queue.setAdminsThatShouldDeclare(rabbitAdmin);
return queue;
}

@Bean
public Binding bindingOneQuene(RabbitAdmin rabbitAdmin,DirectExchange httpMessageNotifyDirectExchange, Queue httpMessageOneQueue) {
Binding binding = BindingBuilder.bind(httpMessageOneQueue).to(httpMessageNotifyDirectExchange).with(ApplicationConstant.HTTP_MESSAGE_ONE_RK);
binding.setAdminsThatShouldDeclare(rabbitAdmin);
return binding;
}

//-------------设置延迟队列--开始--------------------
@Bean
public Queue httpDelayOneQueue() {
//name - 队列名称
//durable - true
//exclusive - false
//autoDelete - false
return QueueBuilder.durable("http.message.dlx.one")
//以下是重点：当变成死信队列时，会转发至 路由为x-dead-letter-exchange及x-dead-letter-routing-key的队列中
.withArgument("x-dead-letter-exchange", ApplicationConstant.HTTP_MESSAGE_EXCHANGE)
.withArgument("x-dead-letter-routing-key", ApplicationConstant.HTTP_MESSAGE_ONE_RK)
.withArgument("x-message-ttl", 1*60*1000)//1分钟 过期时间（单位：毫秒），当过期后 会变成死信队列，之后进行转发
.build();
}
//绑定到交换机上
@Bean
public Binding bindingDelayOneQuene(RabbitAdmin rabbitAdmin, DirectExchange httpMessageNotifyDirectExchange, Queue httpDelayOneQueue) {
Binding binding = BindingBuilder.bind(httpDelayOneQueue).to(httpMessageNotifyDirectExchange).with("delay.one");
binding.setAdminsThatShouldDeclare(rabbitAdmin);
return binding;
}
//-------------设置延迟队列--结束--------------------

//建议将正常的队列和延迟处理的队列分开
//设置监听容器
@Bean("notifyListenerContainer")
public SimpleRabbitListenerContainerFactory httpNotifyListenerContainer() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 手动ack
factory.setConnectionFactory(connectionFactory);
factory.setPrefetchCount(1);
factory.setConcurrentConsumers(concurrency);
return factory;
}

// 设置监听容器
@Bean("delayNotifyListenerContainer")
public SimpleRabbitListenerContainerFactory httpDelayNotifyListenerContainer() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 手动ack
factory.setConnectionFactory(connectionFactory);
factory.setPrefetchCount(1);
factory.setConcurrentConsumers(delayConcurrency);
return factory;
}
}

ApplicationConstant.java

public class ApplicationConstant {

/**
* 发送http通知的 exchange 队列
*/
public static final String HTTP_MESSAGE_EXCHANGE = "http.message.exchange";

/**
* 配置消息队列和路由key值
*/
public static final String HTTP_MESSAGE_START_QUEUE_NAME = "http.message.start";
public static final String HTTP_MESSAGE_START_RK = "rk.start";

public static final String HTTP_MESSAGE_ONE_QUEUE_NAME = "http.message.one";
public static final String HTTP_MESSAGE_ONE_RK = "rk.one";

/**
* 通知队列对应的延迟队列关系，即过期队列之后发送到下一个的队列信息,可以根据实际情况添加，当然也可以根据一定规则自动生成
*/
public static final Map<String,String> delayRefMap = new HashMap<String, String>() {
/**
*
*/
private static final long serialVersionUID = -779823216035682493L;

{
put(HTTP_MESSAGE_START_QUEUE_NAME, "delay.one");
}
};
}

简单来说，就是创建一个正常消息发送队列，用于接收http消息请求的参数，同时进行http请求。同时，创建一个延时队列，设置其

x-dead-letter-exchange

x-dead-letter-routing-key

x-message-ttl

2.创建监听类，用于消息的消费操作，此处使用

@RabbitListener

SimpleMessageListenerContainer

/**
*
* @ClassName   类名：HttpMessagerLister
* @Description 功能说明：http通知消费监听接口
* <p>
* TODO
*</p>
************************************************************************
* @date        创建日期：2019年7月17日
* @author      创建人：oKong
* @version     版本号：V1.0
*<p>
***************************修订记录*************************************
*
*   2019年7月17日   oKong   创建该类功能。
*
***********************************************************************
*</p>
*/
@Component
@Slf4j
public class HttpMessagerLister {

@Autowired
HttpMessagerService messagerService;

@RabbitListener(id = "httpMessageNotifyConsumer", queues = {ApplicationConstant.HTTP_MESSAGE_START_QUEUE_NAME}, containerFactory = "notifyListenerContainer")
public void httpMessageNotifyConsumer(Message message, Channel channel) throws Exception {
doHandler(message, channel);
}

@RabbitListener(id= "httpDelayMessageNotifyConsumer", queues = {
ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME,}, containerFactory = "delayNotifyListenerContainer")
public void httpDelayMessageNotifyConsumer(Message message, Channel channel) throws Exception {
doHandler(message, channel);
}

private void doHandler(Message message, Channel channel) throws Exception {
String body = new String(message.getBody(),"utf-8");
String queue = message.getMessageProperties().getConsumerQueue();
log.info("接收到通知请求：{}，队列名：{}",body, queue);
//消息对象转换
try {
HttpEntity httpNotifyDto = JSONUtil.toBean(body, HttpEntity.class);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//发送通知
messagerService.notify(queue, httpNotifyDto);
} catch(Exception e) {
log.error(e.getMessage());
//ack
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}
}

HttpMessagerService.java

@Component
@Slf4j
public class HttpMessagerService {

@Autowired
AmqpTemplate mqTemplate;

public void notify(String queue,HttpEntity httpEntity) {
//发起请求
log.info("开始发起http请求:{}", httpEntity);
try {
switch(httpEntity.getMethod().toLowerCase()) {
case "POST":
HttpUtil.post(httpEntity.getUrl(), httpEntity.getParams());
break;
case "GET":
default:
HttpUtil.get(httpEntity.getUrl(), httpEntity.getParams());
}
} catch (Exception e) {
//发生异常，放入延迟队列中
String nextRk = ApplicationConstant.delayRefMap.get(queue);
if(ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME.equals(queue)) {
//若已经是最后一个延迟队列的消息队列了，则后续可直接放入数据库中 待后续定时策略进行再次发送
log.warn("http通知已经通知N次失败，进入定时进行发起通知,url={}", httpEntity.getUrl());
} else {
log.warn("http重新发送通知：{}, 通知队列rk为：{}, 原队列：{}", httpEntity.getUrl(), nextRk, queue);
mqTemplate.convertAndSend(ApplicationConstant.HTTP_MESSAGE_EXCHANGE, nextRk, cn.hutool.json.JSONUtil.toJsonStr(httpEntity));
}
}
}
}

3.创建控制层服务（真实场景中，如

SpringCloud

@Slf4j
@RestController
@Api(tags = "http测试接口")
public class HttpDemoController {

@Autowired
AmqpTemplate mqTemplate;

@PostMapping("/send")
@ApiOperation(value="send",notes = "发送http测试")
public String sendHttp(@RequestBody HttpEntity httpEntity) {
//发送http请求
log.info("开始发起http请求，发布异步消息：{}", httpEntity);
mqTemplate.convertAndSend(ApplicationConstant.HTTP_MESSAGE_EXCHANGE, ApplicationConstant.HTTP_MESSAGE_START_RK, cn.hutool.json.JSONUtil.toJsonStr(httpEntity));
return "发送成功：url=" + httpEntity.getUrl();
}
}

4.配置文件添加

RabbitMQ

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

# 通知-消费者线程数 设置大点 大概率是能通知到的
http.notify.concurrency=150
# 延迟队列的消费者线程数 可设置小点
http.notify.delay.concurrency=10

5.编写启动类。

@SpringBootApplication
@Slf4j
public class DelayQueueApplication {

public static void main(String[] args) throws Exception {
SpringApplication.run(DelayQueueApplication.class, args);
log.info("spring-boot-rabbitmq-delay-queue-chapter38服务启动!");
}
}

6.启动服务。使用

swagger

• 正常通知：

2019-07-20 23:52:23.792  INFO 65216 --- [nio-8080-exec-1] c.l.l.s.c.controller.HttpDemoController  : 开始发起http请求，发布异步消息：HttpEntity(url=www.baidu.com, params={a=1}, method=get)
2019-07-20 23:52:23.794  INFO 65216 --- [TaskExecutor-97] c.l.l.s.chapter38.mq.HttpMessagerLister  : 接收到通知请求：{"method":"get","params":{"a":1},"url":"www.baidu.com"}，队列名：http.message.start
2019-07-20 23:52:23.794  INFO 65216 --- [TaskExecutor-97] c.l.l.s.c.service.HttpMessagerService    : 开始发起http请求:HttpEntity(url=www.baidu.com, params={a=1}, method=get)

• 异常通知：访问一个不存在的地址

2019-07-20 23:53:14.699  INFO 65216 --- [nio-8080-exec-4] c.l.l.s.c.controller.HttpDemoController  : 开始发起http请求，发布异步消息：HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:53:14.705  INFO 65216 --- [TaskExecutor-84] c.l.l.s.chapter38.mq.HttpMessagerLister  : 接收到通知请求：{"method":"get","params":{"a":1},"url":"www.baidu.com1"}，队列名：http.message.start
2019-07-20 23:53:14.705  INFO 65216 --- [TaskExecutor-84] c.l.l.s.c.service.HttpMessagerService    : 开始发起http请求:HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:53:14.706  WARN 65216 --- [TaskExecutor-84] c.l.l.s.c.service.HttpMessagerService    : http重新发送通知：www.baidu.com1, 通知队列rk为：delay.one, 原队列：http.message.start

在

RabbitMQ

http.message.dlx.one

http.message.one

在一分钟后，可以看见消息本再次消费了。

2019-07-20 23:54:14.722  INFO 65216 --- [TaskExecutor-16] c.l.l.s.chapter38.mq.HttpMessagerLister  : 接收到通知请求：{"method":"get","params":{"a":1},"url":"www.baidu.com1"}，队列名：http.message.one
2019-07-20 23:54:14.723  INFO 65216 --- [TaskExecutor-16] c.l.l.s.c.service.HttpMessagerService    : 开始发起http请求:HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:54:14.723  WARN 65216 --- [TaskExecutor-16] c.l.l.s.c.service.HttpMessagerService    : http通知已经通知N次失败，进入定时进行发起通知,url=www.baidu.com1

一些最佳实践

RabbitListenerEndpointRegistry

@RabbitListener

@Autowired
RabbitListenerEndpointRegistry registry;

@GetMapping("/set")
@ApiOperation(value = "set", notes = "设置消息监听器的状态")
public String setSimpleMessageListenerContainer(String status) {
if("1".equals(status)) {
registry.getListenerContainer("httpDelayMessageNotifyConsumer").start();
} else {
registry.getListenerContainer("httpDelayMessageNotifyConsumer").stop();
}
return status;
}

这里，只是简单进行演示说明，在真实场景下，可以使用定时器，判断当前是否为高峰期，进而进行动态设置监听器的状态。

参考资料

1. https://www.rabbitmq.com/admin-guide.html
2. https://www.rabbitmq.com/ttl.html

总结

RabbitMQ

redis

最后

SpringBoot

老生常谈

• 个人QQ：

  499452441

• 微信公众号：

  lqdevOps

个人博客：http://blog.lqdev.cn
完整示例：基于RabbitMQ实现消息延迟队列方案
原文地址：https://blog.lqdev.cn/2019/07/21/springboot/chapter-thirty-eight/