Spring boot + RabbitMQ延迟队列实战

一、背景

延时队列顾名思义，即放置在该队列里面的消息是不需要立即消费的，而是等待一段时间之后取出消费。
那么，为什么需要延迟消费呢？我们来看以下的场景:

订单业务：在电商/点餐中，都有下单后 30 分钟内没有付款，就自动取消订单。
短信通知：下单成功后 60s 之后给用户发送短信通知。
失败重试：业务操作失败后，间隔一定的时间进行失败重试。

传统订单处理：

采取定时任务轮训数据库订单，并且批量处理。其弊端也是显而易见的；对服务器、数据库性会有很大的要求，并且当处理大量订单起来会很力不从心,而且实时性也不是特别好。

当然传统的手法还可以再优化一下，即存入订单的时候就算出订单的过期时间插入数据库，设置定时任务查询数据库的时候就只需要查询过期了的订单，然后再做其他的业务操作。

但是通过定时任务来执行，实时性始终不是很好。

二、实现方案

方案一：通过死信队列实现

参考：https://www.cnblogs.com/xmf3628/p/12097101.html

方案二：通过延迟路由插件来实现rabbitmq-delayed-message-exchange

参考：https://www.cnblogs.com/wintercloud/p/10877399.html

这里演示通过插件来实现

三、插件安装

下载地址：
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/v3.8.0

RabbitMQ的有些插件没有集成在初始的安装中，它们需要额外安装，这些文件的后缀为.ez，安装时需要将.ez文件拷贝到安装的插件目录。以下是不同系统中默认安装的插件目录路径：

//查看已安装的插件
rabbitmq-plugins list
//启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
//重启服务
service rabbitmq-server restart
//再次查看，插件是否生效
rabbitmq-plugins list

说明插件安装成功，并成功启用。

四、机制解释

安装插件后会生成新的Exchange类型x-delayed-message，该类型消息支持延迟投递机制,接收到消息后并未立即将消息投递至目标队列中，而是存储在mnesia(一个分布式数据系统)表中，检测消息延迟时间，如达到可投递时间时并将其通过x-delayed-type类型标记的交换机类型投递至目标队列。

五、代码实战

1、依赖

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.44</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</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></dependency>
</dependencies>

2、队列连接信息

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456
spring.rabbitmq.virtual-host=/

3、生产者代码

声明延迟队列和延迟队列的交换器，并建立绑定关系

@Configuration
public class TestDelayQueueConfig {@Beanpublic CustomExchange delayExchange() {Map<String, Object> args = new HashMap<String, Object>();args.put("x-delayed-type", "direct");return new CustomExchange(ExchangeEnum.DELAY_EXCHANGE.getValue(), "x-delayed-message", true, false, args);}/*** 延迟消息队列* @return*/@Beanpublic Queue delayQueue() {return new Queue(QueueEnum.TEST_DELAY.getName(), true);}@Beanpublic Binding deplyBinding() {return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(QueueEnum.TEST_DELAY.getRoutingKey()).noargs();}
}

@Getter
public enum ExchangeEnum {DELAY_EXCHANGE("test.deply.exchange");private String value;ExchangeEnum(String value) {this.value = value;}
}

@Getter
public enum QueueEnum {/*** delay*/TEST_DELAY("test.delay.queue", "delay");/*** 队列名称*/private String name;/*** 队列路由键*/private String routingKey;QueueEnum(String name, String routingKey) {this.name = name;this.routingKey = routingKey;}
}

延迟队列生产者服务：

@Component
@Slf4j
public class DeplyProducer {@Autowiredprivate RabbitTemplate rabbitTemplate;public void send(String msg, int delayTime) {log.info("msg= " + msg + ".delayTime" + delayTime);MessageProperties messageProperties = new MessageProperties();messageProperties.setDelay(delayTime);Message message = new Message(msg.getBytes(), messageProperties);rabbitTemplate.send(ExchangeEnum.DELAY_EXCHANGE.getValue(), QueueEnum.TEST_DELAY.getRoutingKey(), message);}
}

单元测试：

@Test
public void sendDeplyMsgTest() {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");String currentTime = sdf.format(new Date());log.info("发送测试消息的时间：" +currentTime );deplyProducer.send(currentTime + "发送一个测试消息,延迟10秒", 10000);//10秒deplyProducer.send(currentTime + "发送一个测试消息，延迟20秒", 20000);//2秒deplyProducer.send(currentTime + "发送一个测试消息，延迟30秒", 30000);//1秒
}

消费者：

@Component
@RabbitListener(queues = "test.delay.queue")
@Slf4j
public class DeplyConsumer {@RabbitHandlerpublic void onMessage(byte[] message,@Headers Map<String, Object> headers,Channel channel) {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//log.info("收到延时消息时间：" + sdf.format(new Date()) + " Delay sent.");log.info(sdf.format(new Date())+"接收到延时消息:" + new String(message));}
}

执行测试结果：

执行消息发送代码的时候，不会立刻把消息推送到对应的队列中。
只要到了对应的时候，才会将消息推送到队列里面。

可以在执行单元测试的时候，通过管理界面，观察队列中消息的增长：
每隔10s，增加1条。