RabbitMQ-延迟队列
Posted 张铁牛
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ-延迟队列相关的知识,希望对你有一定的参考价值。
1. 简介
我们在上一篇博文中遗留了一个小问题,就是虽然TTL + DLX
能实现延迟队列的功能,但是有两个问题。
首先业务场景为:比如海底捞预约,每个人预约的时间段不一致,有个可能一个小时后,有的可能三个小时等,当快到预约时间点需要给用户进行短信通知。
- 通过给
Queue
设置过期时间的方式不现实,因为很有可能每条记录的过期时间都不一样,不可能设置那么多的Queue
。 - 直接给
Message
设置过期时间,这种方式也不好,因为这种方式是当该消息在队列头部时(消费时),才会单独判断这一消息是否过期。例:现在有两条消息,第一条消息过期时间为30s,而第二条消息过期时间为15s,当过了15秒后,第二条消息不会立即过期,而是要等第一条消息被消费后,第二条消息被消费时,才会判断是否过期,也就是等到第二条消息投往DLX
已经过去45s了。
这也就抛出了本章主题:延迟队列
。
RabbitMQ默认没有提供延迟队列
功能,而是要通过插件提供的x-delayed-message
(延迟交换机)来实现。
延迟队列
:用户可以使用该类型声明一个交换,x-delayed-message
然后使用自定义标头发布消息,x-delay
以毫秒为单位表示消息的延迟时间。消息将在x-delay
毫秒后传递到相应的队列。
2. 安装插件
官方插件地址:https://www.rabbitmq.com/community-plugins.html
找到插件rabbitmq_delayed_message_exchange
,进入GitHub下载本地RabbitMQ对应的插件版本(下载.ez
文件)。
我这里下载的是3.8.9版本,如图:
下载到本地后将文件放置RabbitMQ的plugins目录。
我这里本地是使用docker-compose安装的服务,image
为rabbitmq:3.8.3-management
(虽然版本没对起来,但是测试能用,但是使用3.9的版本会报错,插件安装失败)安装的服务,操作步骤如下:
-
将下载好的文件放置RabbitMQ插件目录
rabbitmq:容器服务名
$ docker cp /Users/ludangxin/Downloads/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez rabbitmq:/opt/rabbitmq/plugins/
-
进入容器
$ docker exec -it rabbitmq /bin/bash
-
查看现有的插件列表
$ rabbitmq-plugins list # 输出部分内容如下 [E*] = 明确启用; e = 隐式启用 [ ] rabbitmq_amqp1_0 3.8.3 [ ] rabbitmq_auth_backend_cache 3.8.3 [ ] rabbitmq_auth_backend_http 3.8.3 [ ] rabbitmq_auth_backend_ldap 3.8.3 [ ] rabbitmq_auth_backend_oauth2 3.8.3 [ ] rabbitmq_auth_mechanism_ssl 3.8.3 [ ] rabbitmq_consistent_hash_exchange 3.8.3 [ ] rabbitmq_event_exchange 3.8.3 [ ] rabbitmq_federation 3.8.3 [ ] rabbitmq_federation_management 3.8.3 [ ] rabbitmq_jms_topic_exchange 3.8.3 [E*] rabbitmq_management 3.8.3 [e*] rabbitmq_management_agent 3.8.3 [ ] rabbitmq_mqtt 3.8.3
-
启用插件
$ rabbitmq-plugins enable rabbitmq_delayed_message_exchange
-
再次查看安装列表就有了
rabbitmq_delayed_message_exchange
安装完毕后登陆RabbitMQ控制台查看,会发现多了个x-delayed-message
类型的Exchange。
3. 实现延迟队列
3.1 引入所需依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</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>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
3.2 application.yaml
spring:
rabbitmq:
host: localhost
port: 5672
# rabbit 默认的虚拟主机
virtual-host: /
# rabbit 用户名密码
username: admin
password: admin123
3.3 RabbitConfig
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
* 延迟队列配置
*
* @author ludangxin
* @date 2021/9/16
*/
@Configuration
public class RabbitDelayedConfig {
public static final String QUEUE_NAME_DELAYED = "DELAY.QUEUE";
public static final String EXCHANGE_NAME_DELAYED = "DELAY.EXCHANGE";
public static final String ROUTING_KEY_DELAYED = "DELAY.#";
@Bean(QUEUE_NAME_DELAYED)
public Queue queue() {
return QueueBuilder.durable(QUEUE_NAME_DELAYED).build();
}
@Bean(EXCHANGE_NAME_DELAYED)
public CustomExchange exchange() {
Map<String, Object> arguments = new HashMap<>(1);
// 在这里声明一个主题类型的延迟队列,当然其他类型的也可以。
arguments.put("x-delayed-type", "topic");
return new CustomExchange(EXCHANGE_NAME_DELAYED, "x-delayed-message", true, false, arguments);
}
@Bean
public Binding bindingNotify(@Qualifier(QUEUE_NAME_DELAYED) Queue queue, @Qualifier(EXCHANGE_NAME_DELAYED) CustomExchange customExchange) {
return BindingBuilder.bind(queue).to(customExchange).with(ROUTING_KEY_DELAYED).noargs();
}
}
3.4 Producer
import com.ldx.rabbitmq.config.RabbitDelayedConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
/**
* 延迟消息生产者
*
* @author ludangxin
* @date 2021/9/9
*/
@Component
public class DelayProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendDelayedMsg(String msg, Integer delay) {
MessageProperties mp = new MessageProperties();
// 设置过期时间
mp.setDelay(delay);
Message message = new Message(msg.getBytes(), mp);
rabbitTemplate.convertAndSend(RabbitDelayedConfig.EXCHANGE_NAME_DELAYED, "DELAY.MSG", message);
}
}
3.5 Consumer
import com.ldx.rabbitmq.config.RabbitDelayedConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
/**
* 延迟消息消费者
*
* @author ludangxin
* @date 2021/9/9
*/
@Slf4j
@Component
public class DelayConsumer {
@RabbitListener(queues = {RabbitDelayedConfig.QUEUE_NAME_DELAYED})
public void delayQueue(Message message){
log.info(new String(message.getBody()) + ",结束时间为:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
}
}
3.6 测试代码
@Autowired
private DelayProducer delayProducer;
@Test
@SneakyThrows
public void sendDelayedMsg() {
for(int i = 16; i >= 10; i --) {
String msg = "我将在" + i + "s后过期,开始时间为:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
delayProducer.sendDelayedMsg(msg,i * 1000);
}
// 使进程阻塞,方便Consumer监听输出Message
System.in.read();
}
3.7 启动测试
启动测试代码,输出内容如下:
从日志内容可以看出,消息存活了30s,符合预期。
2021-09-16 23:40:10.806 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我将在10s后过期,开始时间为:2021-09-16 23:40:00,结束时间为:2021-09-16 23:40:10
2021-09-16 23:40:11.792 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我将在11s后过期,开始时间为:2021-09-16 23:40:00,结束时间为:2021-09-16 23:40:11
2021-09-16 23:40:12.791 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我将在12s后过期,开始时间为:2021-09-16 23:40:00,结束时间为:2021-09-16 23:40:12
2021-09-16 23:40:13.791 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我将在13s后过期,开始时间为:2021-09-16 23:40:00,结束时间为:2021-09-16 23:40:13
2021-09-16 23:40:14.788 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我将在14s后过期,开始时间为:2021-09-16 23:40:00,结束时间为:2021-09-16 23:40:14
2021-09-16 23:40:15.785 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我将在15s后过期,开始时间为:2021-09-16 23:40:00,结束时间为:2021-09-16 23:40:15
2021-09-16 23:40:16.785 INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer : 我将在16s后过期,开始时间为:2021-09-16 23:40:00,结束时间为:2021-09-16 23:40:16
以上是关于RabbitMQ-延迟队列的主要内容,如果未能解决你的问题,请参考以下文章
SpringBoot集成RabbitMQ之死信队列限流队列延迟队列(第四节)