目录
- 业务场景
- 传统轮询
- 消息队列
- 完整版
- 默认情况
- TTL（消息的有效期）
- TTL 的设置有两种不同的方式
- 单条消息过期
- 队列消息过期
- 特殊情况
- 死信队列
- 概述
- 应用场景
- 产生原因
- 原理图
- 死信交换机
- 死信队列
- 实现一下
- 延迟队列
- 背景
- 定时任务？
- 延迟队列实现思路
- 代码
业务场景
以电商抢购为例，用户下单后30分钟内未完成支付，则取消订单
传统轮询
消息队列
完整版
默认情况
- 默认情况下，消息是不会过期的，也就是我们平日里在消息发送时，如果不设置任何消息过期的相关参数，那么消息是不会过期的，
- 即使消息没被消费掉，也会一直存储在队列中。
TTL（消息的有效期）
- TTL（Time-To-Live），消息存活的时间，即消息的有效期。如果我们希望消息能够有一个存活时间，那么我们可以通过设置 TTL 来实现这一需求。
- 如果消息的存活时间超过了 TTL 并且还没有被消息，此时消息就会变成死信
TTL 的设置有两种不同的方式
- 在声明队列的时候，我们可以在队列属性中设置消息的有效期，这样所有进入该队列的消息都会有一个相同的有效期。
- 在发送消息的时候设置消息的有效期，这样不同的消息就具有不同的有效期。
- 如果两个都设置了就以时间短的为准。
当我们设置了消息有效期后，消息过期了就会被从队列中删除了（进入到死信队列），但是两种方式对应的删除时机有一些差异
4. 对于第一种方式，当消息队列设置过期时间的时候，那么消息过期了就会被删除，因为消息进入 RabbitMQ 后是存在一个消息队列中，队列的头部是最早要过期的消息，所以 RabbitMQ 只需要一个定时任务，从头部开始扫描是否有过期消息，有的话就直接删除。
5. 对于第二种方式，当消息过期后并不会立马被删除，而是当消息要投递给消费者的时候才会去删除，因为第二种方式，每条消息的过期时间都不一样，想要知道哪条消息过期，必须要遍历队列中的所有消息才能实现，当消息比较多时这样就比较耗费性能，因此对于第二种方式，当消息要投递给消费者的时候才去删除
单条消息过期
-
依赖
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </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> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> -
配置
server: port: 8888 spring: rabbitmq: host: 192.168.29.200 port: 5672 username: admin password: admin virtual-host: / -
配置队列
@Configuration public class QueueConfig { //首先配置一个消息队列，new 一个 Queue：第一个参数是消息队列的名字；第二个参数表示消息是否持久化；第三个参数表示消息队列是否排他，一般我们都是设置为 false，即不排他；第四个参数表示如果该队列没有任何订阅的消费者的话，该队列会被自动删除，一般适用于临时队列。 //配置一个 DirectExchange 交换机。 //将交换机和队列绑定到一起。 public static final String MQ_QUEUE_DEMO = "mq_queue_demo"; public static final String MQ_EXCHANGE_DEMO = "mq_exchange_demo"; public static final String HELLO_ROUTING_KEY = "hello_routing_key"; @Bean public Queue queue() { return new Queue(MQ_QUEUE_DEMO, true, false, false); } @Bean public DirectExchange directExchange() { return new DirectExchange(MQ_EXCHANGE_DEMO, true, false); } @Bean public Binding binding() { return BindingBuilder.bind(queue()) .to(directExchange()) .with(HELLO_ROUTING_KEY); } }
关于排他性
- 如果设置为 true，则该消息队列只有创建它的 Connection 才能访问，其他的 Connection 都不能访问该消息队列
- 如果试图在不同的连接中重新声明或者访问排他性队列，那么系统会报一个资源被锁定的错误。
- 另一方面，对于排他性队列而言，当连接断掉的时候，该消息队列也会自动删除（无论该队列是否被声明为持久性队列都会被删除）
- 生产消息发送
@RestController public class SendController { @Resource private RabbitTemplate rabbitTemplate;; @RequestMapping("/send") public String send() { //在创建 Message 对象的时候我们可以设置消息的过期时间，这里设置消息的过期时间为 10 秒 Message message = MessageBuilder.withBody("hello rabbitmq".getBytes()) .setExpiration("10000") .build(); rabbitTemplate.convertAndSend(QueueConfig.MQ_QUEUE_DEMO, message); return "send success"; } } - 启动项目，进行消息发送测试。当消息发送成功之后，由于没有消费者，所以这条消息并不会被消费。打开 RabbitMQ 管理页面，点击到 Queues 选项卡，10s 之后，会发现消息已经不见了
单条消息设置过期时间，就是在消息发送的时候设置一下消息有效期即可。
队列消息过期
-
给队列设置消息过期时间：配置类修改队列设置
@Bean public Queue queue() { Map<String, Object> args = new HashMap<>(); args.put("x-message-ttl", 10000); return new Queue(MQ_QUEUE_DEMO, true, false, false, args); } -
修改消息的发送逻辑
@RestController public class SendController { @Resource private RabbitTemplate rabbitTemplate;; @RequestMapping("/send") public String send() { //在创建 Message 对象的时候我们可以设置消息的过期时间，这里设置消息的过期时间为 10 秒 /*Message message = MessageBuilder.withBody("hello rabbitmq".getBytes()) .setExpiration("10000") .build(); rabbitTemplate.convertAndSend(QueueConfig.MQ_QUEUE_DEMO, message);*/ Message message = MessageBuilder.withBody("hello rabbitmq".getBytes()) .build(); rabbitTemplate.convertAndSend(QueueConfig.MQ_QUEUE_DEMO, message); return "send success"; } } -
启动项目，发送一条消息进行测试。查看 RabbitMQ 管理页面
可以看到，消息队列的 Features 属性为 D 和 TTL，D 表示消息队列中消息持久化，TTL 则表示消息会过期。10s 之后刷新页面，发现消息数量已经恢复为 0。
这就是给消息队列设置消息过期时间，一旦设置了，所有进入到该队列的消息都有一个过期时间了
特殊情况
- 还有一种特殊情况，就是将消息的过期时间 TTL 设置为 0，这表示如果消息不能立马消费则会被立即丢掉，这个特性可以部分替代 RabbitMQ3.0 以前支持的 immediate 参数
- 之所以所部分代替，是因为 immediate 参数在投递失败会有 basic.return 方法将消息体返回（这个功能可以利用死信队列来实现）
死信队列
概述
- producer 将消息投递到 broker 或者直接到 queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，所有的死信就组成了死信队列。
应用场景
- 为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。
产生原因
- 消息 TTL 过期(单位)
- 队列达到最大长度(队列满了，无法再添加数据到 mq 中)
- 消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false
原理图
- 问题：被删除的消息去哪了？真的被删除了吗？
死信交换机
-
死信交换机，Dead-Letter-Exchange 即 DLX。
-
死信交换机用来接收死信消息（Dead Message）的，那什么是死信消息呢？一般消息变成死信消息有如下几种情况：
- 消息被拒绝(Basic.Reject/Basic.Nack) ，井且设置requeue 参数为false
- 消息过期
- 队列达到最大长度0
-
当消息在一个队列中变成了死信消息后，此时就会被发送到 DLX，绑定 DLX 的消息队列则称为死信队列。
-
DLX 本质上也是一个普普通通的交换机，我们可以为任意队列指定 DLX，当该队列中存在死信时，RabbitMQ 就会自动的将这个死信发布到 DLX 上去，进而被路由到另一个绑定了 DLX 的队列上（即死信队列）。
死信队列
- 绑定了死信交换机的队列就是死信队列
实现一下
-
创建一个死信交换机，一个死信队列，再将死信交换机和死信队列绑定到一起
-
为消息队列配置死信交换机
x-dead-letter-exchange：配置死信交换机。
x-dead-letter-routing-key：配置死信 routing_key。将来发送到这个消息队列上的消息，如果发生了 nack、reject 或者过期等问题，就会被发送到 DLX 上，进而进入到与 DLX 绑定的消息队列上
@Configuration public class RabbitConfig { public static final String DLX_EXCHANGE_NAME = "dlx_exchange_name"; public static final String DLX_QUEUE_NAME = "dlx_queue_name"; public static final String DLX_ROUTING_KEY = "dlx_routing_key"; public static final String MQ_QUEUE_DEMO = "mq_queue_demo"; public static final String MQ_EXCHANGE_NAME = "mq_exchange_name"; public static final String MQ_ROUTING_KEY = "mq_routing_key"; /** * 配置死信交换机 * @return */ @Bean public DirectExchange dlxDirectExchange() { return new DirectExchange(DLX_EXCHANGE_NAME, true, false); } /** * 配置死信队列 * @return */ @Bean public Queue dlxQueue() { return new Queue(DLX_QUEUE_NAME); } /** * 绑定死信队列和死信交换机 * @return */ @Bean public Binding dlxBinding() { return BindingBuilder.bind(dlxQueue()) .to(dlxDirectExchange()) .with(DLX_ROUTING_KEY); } @Bean public Queue nomarlQueue() { Map<String, Object> args = new HashMap<>(); //设置消息过期时间 args.put("x-message-ttl", 0); //设置死信交换机 args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME); //设置死信 routing_key args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY); return new Queue(MQ_QUEUE_DEMO, true, false, false, args); } @Bean public DirectExchange nomarlDirectExchange() { return new DirectExchange(MQ_EXCHANGE_NAME, true, false); } /** * * @return */ @Bean public Binding nomarlBinding() { return BindingBuilder.bind(nomarlQueue()) .to(nomarlDirectExchange()) .with(MQ_ROUTING_KEY); } } -
消费
@Component public class DlxConsumer { @RabbitListener(queues = RabbitConfig.DLX_QUEUE_NAME) public void dlxHandle(String msg) { System.out.println("dlx msg = " + msg); } } -
发送消息
@RestController public class SendController { @Resource private RabbitTemplate rabbitTemplate;; @RequestMapping("/send") public String send() { rabbitTemplate.convertAndSend(RabbitConfig.MQ_EXCHANGE_NAME, RabbitConfig.MQ_ROUTING_KEY, "hello rabbitmq"); return "send success"; } }
延迟队列
背景
- 在电商项目中，当用户下单之后，一般需要 15 分钟之内或者 30 分钟之内付款，否则订单就会进入异常处理逻辑中，被取消，那么进入到异常处理逻辑中，就可以当成是一个延迟队列。
- 公司的会议预定系统，在会议预定成功后，会在会议开始前半小时通知所有预定该会议的用户。
- 工单超过 24 小时未处理，则自动提醒相关责任人。
- 用户下单外卖以后，距离超时时间还有 10 分钟时提醒外卖小哥即将超时。
…
定时任务？
- 如果项目中只有一个这样的延迟队列的场景，那么搞个定时任务似乎也可以
- 但是如果项目中有很多这样的场景，那么定时任务很明显就不是最佳方案了，我们可以通过延迟队列来实现一个通用的解决方案
延迟队列实现思路
延迟队列实现的思路就是上文提到的 DLX（死信交换机）+TTL（消息超时时间），我们可以把死信队列就当成延迟队列
- 假如一条消息需要延迟 30 分钟执行，我们就设置这条消息的有效期为 30 分钟，
- 同时为这条消息配置死信交换机和死信 routing_key，并且不为这个消息队列设置消费者
- 那么 30 分钟后，这条消息由于没有被消费者消费而进入死信队列，
- 此时我们有一个消费者就在“蹲点”这个死信队列，消息一进入死信队列，就立马被消费了
代码
- 其实区别就是把上面案例中普通队列nomarlQueue中的x-message-ttl设置一个过期时间即可，0表示立即过期，15000表示15秒后过期
@Bean
public Queue nomarlQueue() {
Map<String, Object> args = new HashMap<>();
//设置消息过期时间
//args.put("x-message-ttl", 0);
args.put("x-message-ttl", 15000);//15秒
//设置死信交换机
args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
//设置死信 routing_key
args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
return new Queue(MQ_QUEUE_DEMO, true, false, false, args);
}
