业务场景：
- 延迟发送短信
- 用户下单，若15分钟内用户未支付，取消订单
- 预约工作会议，20分钟后通知所有参会人员
…
实现方式1：死信交换机
死信概念：当一个队列中的消息满足下列情况之一，就会变成死信
- 消费者使用basic.reject或basic.nack声明消费失败且消息的requeue（重新入队）参数设置为false
- 消息超时未消费
- 队列里的消息堆积满了，最早未被消费的消息就可能成为死信
如果该队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机 （Dead Letter Exchange，简称DLX)。给队列设置dead-letter-routing-key属性，设置死信交换机与死信队列的RoutingKey
TTL（Time To Live）是指消息的存活时间
RabbitMQ可以对队列和消息分别设置TTL
代码实现
添加依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
rabbitmq配置
spring:
rabbitmq:
host: 127.0.0.1 # rabbitMQ的ip地址
port: 5672 # 端口
username: guest
password: guest
virtual-host: /
死信交换机配置
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class TTLMessageConfig {
@Bean
public DirectExchange ttlDirectExchange(){
return new DirectExchange("ttl.direct");
}
@Bean
public Queue ttlQueue(){
return QueueBuilder
// 创建并持久化队列
.durable("ttl.queue")
// 设置队列的消息存活时间
.ttl(10000)
// 给该队列绑定死信交换机并指定routing-key路由到死信队列
.deadLetterExchange("dl.direct")
.deadLetterRoutingKey("dl")
.build();
}
@Bean
public Binding ttlBinding(){
// 将ttl.queue队列绑定到ttl.direct
return BindingBuilder.bind(ttlQueue()).to(ttlDirectExchange()).with("ttl");
}
}
监听队列
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class SpringRabbitListener {
// 监听队列
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "dl.queue", durable = "true"),
exchange = @Exchange(name = "dl.direct"),
key = "dl"
))
public void listenDlQueue(String msg) {
log.info("消费者接收到了dl.queue的延迟消息");
}
}
测试
@Test
public void testTTLMessage() {
// 1.准备消息
Message message = MessageBuilder
.withBody("hello, ttl messsage".getBytes(StandardCharsets.UTF_8))
//消息持久化(默认)
.setDeliveryMode(MessageDeliveryMode.PERSISTENT)
// 消息过时时间(当消息的过时时间与队列的过时时间都设置时，值小的生效)
.setExpiration("15000")
.build();
// 2.发送消息
rabbitTemplate.convertAndSend("ttl.direct", "ttl", message);
// 3.记录日志
log.info("消息已经成功发送！");
}
实现方式2：使用插件实现延迟队列
GitHub地址
1、下载完成之后进行解压，此处推荐bandzip进行解压，并且将解压之后的文件夹放到rabbitmq的安装目录下的plugins目录下
2、进入到sbin目录下使用cmd执行以下指令来启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
3、然后重启rabbitmq服务
可以看到交换机类型添加了x-delayed-message类型，此类型是基于fanout、direct、topic类型的
监听队列
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "delay.queue", durable = "true"),
// delayed属性指定该交换机为延迟交换机
exchange = @Exchange(name = "delay.direct", delayed = "true"),
key = "delay"
))
public void listenDelayExchange(String msg) {
log.info("消费者接收到了delay.queue的延迟消息");
}
测试
@Test
public void testSendDelayMessage() throws InterruptedException {
// 1.准备消息
Message message = MessageBuilder
.withBody("hello, ttl messsage".getBytes(StandardCharsets.UTF_8))
.setDeliveryMode(MessageDeliveryMode.PERSISTENT)
// x-delay设置延时时间
.setHeader("x-delay", 15000)
.build();
// 2.准备CorrelationData
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 3.发送消息
rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);
log.info("发送消息成功");
}
