SpringBoot + RabbitMQ 做延时队列

グ〞夜微涼 ~

一、前言

延迟队列的使用场景：
　　1.未按时支付的订单，30分钟过期之后取消订单；
　　2.给活跃度比较低的用户间隔N天之后推送消息，提高活跃度；
　　3.过1分钟给新注册会员的用户，发送注册邮件等。

实现延迟队列的方式有两种：

通过消息过期后进入死信交换器，再由交换器转发到延迟消费队列，实现延迟功能；
使用rabbitmq-delayed-message-exchange插件实现延迟功能；
注意： 延迟插件rabbitmq-delayed-message-exchange是在RabbitMQ 3.5.7及以上的版本才支持的，依赖Erlang/OPT 18.0及以上运行环境。
由于使用死信交换器相对曲折，本文重点介绍第二种方式，使用rabbitmq-delayed-message-exchange插件完成延迟队列的功能。

AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能，但是我们可以通过RabbitMQ的两个特性来曲线实现延迟队列：

特性一：Time To Live(TTL)

RabbitMQ可以针对Queue设置x-expires 或者 针对Message设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)
RabbitMQ针对队列中的消息过期时间有两种方法可以设置。
A: 通过队列属性设置，队列中所有消息都有相同的过期时间。
B: 对消息进行单独设置，每条消息TTL可以不同。

如果同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就成为dead letter

特性二：Dead Letter Exchanges（DLX）

RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由转发到指定的队列。
x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange
x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送
队列出现dead letter的情况有：
消息或者队列的TTL过期
队列达到最大长度
消息被消费端拒绝（basic.reject or basic.nack）并且requeue=false

二、安装延迟插件

1.1 下载插件

下载和安装详细步骤

SpringBoot整合RabbitMQ

创建一个springBoot项目

在 pom.xml 中添加 spring-boot-starter-amqp的依赖

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>

在 application.yml文件中配置rabbitmq相关内容

spring:
　rabbitmq:
　 　host: 127.0.0.1
　 　port: 5672
　 　username: guest
　 　password: guest
　 　listener:
　 　direct:
　 　 　acknowledge-mode: manual
　 　simple:
　 　 　acknowledge-mode: manual

具体编码实现

1.配置队列

package com.hmg.rabbitmq.config;

import lombok.extern.log4j.Log4j2;
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;

/**
* @title rabbitmq配置类
* @auther 吊炸天
* @date 2019/12/18 21:28
*/
@Configuration
@Log4j2
public class DelayRabbitConfig {

/**
* 延迟队列 TTL 名称
*/
private static final String ORDER_DELAY_QUEUE = "user.order.delay.queue";
/**
* DLX，dead letter发送到的 exchange
* 延时消息就是发送到该交换机的
*/
public static final String ORDER_DELAY_EXCHANGE = "user.order.delay.exchange";
/**
* routing key 名称
* 具体消息发送在该 routingKey 的
*/
public static final String ORDER_DELAY_ROUTING_KEY = "order_delay";

public static final String ORDER_QUEUE_NAME = "user.order.queue";
public static final String ORDER_EXCHANGE_NAME = "user.order.exchange";
public static final String ORDER_ROUTING_KEY = "order";

/**
* 延迟队列配置
* <p>
* 1、params.put("x-message-ttl", 5 * 1000);
* 第一种方式是直接设置 Queue 延迟时间 但如果直接给队列设置过期时间,这种做法不是很灵活,（当然二者是兼容的,默认是时间小的优先）
* 2、rabbitTemplate.convertAndSend(book, message -> {
* message.getMessageProperties().setExpiration(2 * 1000 + "");
* return message;
* });
* 第二种就是每次发送消息动态设置延迟时间,这样我们可以灵活控制
**/
@Bean
public Queue delayOrderQueue() {
Map<String, Object> params = new HashMap<>();
// x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，
params.put("x-dead-letter-exchange", ORDER_EXCHANGE_NAME);
// x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
params.put("x-dead-letter-routing-key", ORDER_ROUTING_KEY);
return new Queue(ORDER_DELAY_QUEUE, true, false, false, params);
}

/**
* 需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。
* 这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 “dog”，则只有被标记为“dog”的消息才被转发，
* 不会转发dog.puppy，也不会转发dog.guard，只会转发dog。
*
* @return DirectExchange
*/
@Bean
public DirectExchange orderDelayExchange() {
return new DirectExchange(ORDER_DELAY_EXCHANGE);
}

@Bean
public Binding dlxBinding() {
return BindingBuilder.bind(delayOrderQueue()).to(orderDelayExchange()).with(ORDER_DELAY_ROUTING_KEY);
}

@Bean
public Queue orderQueue() {
return new Queue(ORDER_QUEUE_NAME, true);
}

/**
* 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。
* 符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.*” 只会匹配到“audit.irs”。
**/
@Bean
public TopicExchange orderTopicExchange() {
return new TopicExchange(ORDER_EXCHANGE_NAME);
}

@Bean
public Binding orderBinding() {
// TODO 如果要让延迟队列之间有关联,这里的 routingKey 和 绑定的交换机很关键
return BindingBuilder.bind(orderQueue()).to(orderTopicExchange()).with(ORDER_ROUTING_KEY);
}

}

2.创建一个Order实体类

package com.hmg.rabbitmq.entity;

import lombok.Data;

import java.io.Serializable;

/**
* @auther 吊炸天
* @date 2019/12/18 21:43
*/
@Data
public class Order implements Serializable {

private static final long serialVersionUID = -2221214252163879885L;

/**
* 订单id
*/
private String orderId;

/**
* 订单状态 0：未支付，1：已支付，2：订单已取消
*/
private Integer orderStatus;

/**
* 订单名字
*/
private String orderName;
}

3.接收者

package com.hmg.rabbitmq.config;

import com.hmg.rabbitmq.entity.Order;

import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
* @auther 吊炸天
* @date 2019/12/18 21:46
*/
@Component
@Log4j2
public class DelayReceiver {

@RabbitListener(queues = {DelayRabbitConfig.ORDER_QUEUE_NAME})
public void orderDelayQueue(Order order, Message message, Channel channel) {
log.info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
log.info("【orderDelayQueue 监听的消息】 - 【消费时间】 - [{}]- 【订单内容】 - [{}]",  new Date(), order.toString());
if(order.getOrderStatus() == 0) {
order.setOrderStatus(2);
log.info("【该订单未支付，取消订单】" + order.toString());
} else if(order.getOrderStatus() == 1) {
log.info("【该订单已完成支付】");
} else if(order.getOrderStatus() == 2) {
log.info("【该订单已取消】");
}
log.info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
}

}

4.发送者

package com.hmg.rabbitmq.config;

import com.hmg.rabbitmq.entity.Order;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
* @auther 吊炸天
* @date 2019/12/18 21:48
*/
@Component
@Log4j2
public class DelaySender {

@Autowired
private AmqpTemplate amqpTemplate;

public void sendDelay(Order order) {

log.info("【订单生成时间】" + new Date().toString() + "【1分钟后检查订单是否已经支付】" + order.toString());
this.amqpTemplate.convertAndSend(DelayRabbitConfig.ORDER_DELAY_EXCHANGE,
DelayRabbitConfig.ORDER_DELAY_ROUTING_KEY, order, message -> {
/**
* 如果配置了 params.put("x-message-ttl", 5 * 1000);
* 那么这一句也可以省略,
* 具体根据业务需要是声明 Queue 的时候就指定好延迟时间还是在发送自己控制时间
*/
message.getMessageProperties().setExpiration(1 * 1000 * 60 + "");
return message;
});
}
}

5.测试，访问http://localhost:8080/sendDelay，查看日志输出

package com.hmg.rabbitmq.controller;

import com.hmg.rabbitmq.config.DelaySender;
import com.hmg.rabbitmq.entity.Order;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
* @auther 吊炸天
* @date 2019/12/18 21:56
*/
@RestController
public class TestController {

@Autowired
private DelaySender delaySender;

@GetMapping("/sendDelay")
public Object sendDelay() {
Order order1 = new Order();
order1.setOrderStatus(0);
order1.setOrderId("13147747");
order1.setOrderName("魅族16plus");

Order order2 = new Order();
order2.setOrderStatus(1);
order2.setOrderId("68363685");
order2.setOrderName("魅族16s");

delaySender.sendDelay(order1);
delaySender.sendDelay(order2);
return "ok";
}
}

6.测试效果

• 点赞 2
• 收藏
• 分享
• 文章举报