根本原理rabbitmq支持过期时间.
给队列设置过期时间15秒:那么进入此队列的所有消息都只会存活15秒
先看只给队列设置过期时间,其他不设置.
定义队列和交换机,注意给队列加上
过期时间属性
过期时间不是指的队列过期,而是队列里的消息.
import java.util.HashMap;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
@Bean
public Queue myQueue() {
HashMap<String,Object> hashMap = new HashMap<>();
hashMap.put("x-message-ttl", 15000);//定义队列里消息存活时间15秒
return QueueBuilder.durable("myQueue").withArguments(hashMap).build();
}
@Bean
public Exchange myExchange() {
return new DirectExchange("myExchange");
}
@Bean
public Binding myBinding(
Queue myQueue,
Exchange myExchange)
{
return BindingBuilder
.bind(myQueue)
.to(myExchange)
.with("myKey")
.noargs();
}
}
写一个控制器,作为生产者,向队列推入消息
package com.spring_boot.controller;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.spring_boot.config.RabbitMQConfig;
@RestController
public class MQTestController {
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("deadMessage")
public String deadMessage() {
rabbitTemplate.convertAndSend("myExchange", "myKey", "这是消息");
return "done";
}
}
访问控制器,查看队列后台
图片.png
注意,我们并没有设置消费者,过了15秒,消息超时直接被队列丢弃了.
以下个人的理解,有错误的地方欢迎指出
到期被丢弃的消息就是
死信.rabbitMQ有一种机制可以把死信投递到指定的交换机.再由交换机转发到合适的队列.
用来存放死信的队列就叫做死信队列.
用来把死信转发给队列的交换机叫做死信交换机
接下来尝试把过期的消息投递到死信队列
- 首先应该有一个
交换机来接收死信,还要有一个队列来保存死信
在配置类中定义死信交换机和死信队列.(和普通的队列一样,叫法不同只是为了便于区分)
#RabbitConfig.java
/*
死信队列名为 deadQueue
死信交换机名为 deadExchange
死信路由键名为 deadKey
*/
@Bean//死信队列
public Queue deadQueue() {
return new Queue("deadQueue");
}
@Bean//死信交换机
public Exchange deadExchange() {
return new DirectExchange("deadExchange");
}
@Bean//绑定
public Binding deadBinding(Queue deadQueue,Exchange deadExchange) {
return BindingBuilder
.bind(deadQueue)
.to(deadExchange)
.with("deadKey")
.noargs();
}
- 我们必须告诉rabbitMQ把过期的消息投递给
死信交换机
回到最开始设置过期时间的地方,在队列定义时增加两个属性
指明,死信应该以deadKey为路由键,投递到deadExchange交换机
@Bean
public Queue myQueue() {
HashMap<String,Object> hashMap = new HashMap<>();
hashMap.put("x-message-ttl", 15000);//定义消息存活时间15秒
hashMap.put("x-dead-letter-exchange", "deadExchange");//设置此队列产生的死信会投递给哪个交换机
hashMap.put("x-dead-letter-routing-key", "deadKey");//设置死信投递时使用的路由键
return QueueBuilder.durable("myQueue").withArguments(hashMap).build();
}
- 手动在后台删掉之前生成的队列,和交换机重新运行项目,观察结果
访问控制器, 生产者推送消息到myQueue由于图片.pngmyQueue设置了TTL=15000,15秒后,消息过期变成死信,被投递给死信交换机最终进入deadQueue.图片.png
总结:
转换一下思维,商城的请在15分钟内支付应该就是这道理
- 用户下单时,订单信息写入数据库,状态为待付款.
- 订单ID放入myQueue,15分钟过期.
(在这15分钟里,如果用户付款了,肯定会有回调函数把数据库里对应的支付状态改成已付款,订单ID继续留在队列里就好了) - 15分钟一到,订单ID就会进入
死信队列
那么只需要定义一个消费者,去消费死信队列里的消息.
每取出一个订单ID就去数据库判断这个订单的支付状态,如果仍是未付款,就给他取消掉.支付成功的不用管
实验,给
死信队列配置一个消费者
package com.spring_boot.consumer;
import java.text.SimpleDateFormat;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class Receiver {
@RabbitHandler
@RabbitListener(queues = {"deadQueue"})
public void cancelOrder(Message message) {
//去数据库查询订单支付状态
boolean paySuccess = false;
if (!paySuccess) {
System.out.println(new String(message.getBody()));
System.out.println("订单超时,取消订单"+new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(System.currentTimeMillis()));
}
}
}
图片.png
)
