rabbitmq 实现延迟队列
什么是延迟队列
延迟队列存储的对象肯定是对应的延时消息，所谓”延时消息”是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。
比如以下场景：在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将进行一场处理。这是就可以使用延时队列将订单信息发送到延时队列。
RabbitMQ如何实现迟队列
AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。
但是我们可以通过RabbitMQ的两个特性来曲线实现延迟队列：
- 特性1、Time To Live(TTL)(官方文档)
RabbitMQ可以针对Queue设置x-expires 或者 针对Message设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)
RabbitMQ针对队列中的消息过期时间有两种方法可以设置。
A: 通过队列属性设置，队列中所有消息都有相同的过期时间。
B: 对消息进行单独设置，每条消息TTL可以不同。
如果同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就成为dead letter
- 特性2、Dead Letter Exchanges（DLX）(官方文档)
RabbitMQ的Queue可以配置x-dead-letter-exchange和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由转发到指定的队列。
x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange
x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送
队列出现dead letter的情况有：
消息或者队列的TTL过期
队列达到最大长度
消息被消费端拒绝（basic.reject or basic.nack）并且requeue=false
综合上述两个特性，设置了TTL规则之后当消息在一个队列中变成死信时，利用DLX特性它能被重新转发到另一个Exchange或者Routing Key，这时候消息就可以重新被消费了。
具体实现思路
我们可以创建两个队列
1. 订单队列
2. 超时订单队列
订单队列配置转发规则（如果没有任何规则，则直接丢弃死信），当队列中消息超时的时候自动转发到超时订单队列中
Spring boot 实现
1、引入spring-rabbit
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
</dependency>
2、配置application.properties
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.virtual-host=/
3、配置订单队列，订单超时队列，并配置相应的exchange和routingKey
@Configuration
public class OrderQueueConfig {
/**
* 订单队列
*/
public static final String ORDER_QUEUE = "woniu.order.queue";
/**
* 超时订单队列
*/
public static final String ORDER_EXCHANGE = "woniu.order.exchage";
/**
* 订单exchange
*/
public static final String ORDER_TIMEOUT_QUEUE = "woniu.order.timeout.queue";
/**
* 订单exchange
*/
@Bean
public DirectExchange orderExchange(){
return new DirectExchange(ORDER_EXCHANGE,true,false,null);
}
/**
* 订单队列
*/
@Bean
public Queue orderQueue() {
// 设置超时转发策略 超时后消息会通过x-dead-letter-exchange 转发到x-dead-letter-routing-key绑定的队列中
Map<String, Object> arguments = new HashMap<>(2);
arguments.put("x-dead-letter-exchange", ORDER_EXCHANGE);
arguments.put("x-dead-letter-routing-key", ORDER_TIMEOUT_QUEUE);
Queue queue = new Queue(ORDER_QUEUE,true,false,false,arguments);
return queue;
}
/**
* 超时订单队列
* @return
*/
@Bean
public Queue orderTimeoutQueue() {
Queue queue = new Queue(ORDER_TIMEOUT_QUEUE,true,false,false);
return queue;
}
/**
* 订单队列绑定exchange
* @return
*/
@Bean
public Binding orderQueueBinding() {
return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(ORDER_QUEUE);
}
/**
* 超时订单队列绑定exchange
* @return
*/
@Bean
public Binding orderTimeoutQueueBinding() {
return BindingBuilder.bind(orderTimeoutQueue()).to(orderExchange()).with(ORDER_TIMEOUT_QUEUE);
}
}
4、创建生产者，并在发送消息的时候指定超时时间
@Service
public class OrderProducer {
@Autowired
private AmqpTemplate template;
public void sendOrder(String orderId) {
template.convertAndSend(OrderQueueConfig.ORDER_EXCHANGE, OrderQueueConfig.ORDER_QUEUE, orderId, message -> {
// 设置超时时间 3000ms
message.getMessageProperties().setExpiration("3000");
return message;
});
}
}
5、创建消费者，消费者消费订单超时队列
@Service
public class OrderConsumer {
@RabbitHandler
@RabbitListener(queues = OrderQueueConfig.ORDER_TIMEOUT_QUEUE,concurrency="4-10")
public void consumeTimeOutQueue(@Payload String orderId){
System.out.println("接收到消息的时间"+System.currentTimeMillis());
System.out.println("消息ID="+orderId);
}
}
6、测试
public String sendOrder(String orderId){
System.out.println("消息发送时间:"+System.currentTimeMillis());
orderProducer.sendOrder(orderId);
return "sendOrder";
}
7、结果
消息发送时间 :1538270251066
接收到消息的时间:1538270254123
消息ID=hello
完！
网友评论