rabbitmq延迟队列的使用
1、场景:
1.定时发布文章
2.秒杀之后,给30分钟时间进行支付,如果30分钟后,没有支付,订单取消。
3.预约餐厅,提前半个小时发短信通知用户。
A -> 13:00 17:00 16:30 延迟时间: 730 * 60 * 1000
B -> 11:00 18:00 17:30 延迟时间: 1330 * 60 * 1000
C -> 8:00 14:00 13:30 延迟时间: 11*30 * 60 * 1000
第一种方式:创建具有超时功能且绑定死信交换机的消息队列
@Bean
public Queue directQueueLong(){
return QueueBuilder.durable("业务队列名称")
.deadLetterExchange("死信交换机名称")
.deadLetterRoutingKey("死信队列 RoutingKey")
.ttl(20000) // 消息停留时间
//.maxLength(500)
.build();
}
监听死信队列,即可处理超时的消息队列
缺点:
上述实现方式中,ttl延时队列中所有的消息超时时间都是一样的,如果不同消息想设置不一样的超时时间,就需要建立多个不同超时时间的消息队列,比较麻烦,且不利于维护。
第二种方式:创建通用延时消息
rabbitTemplate.convertAndSend("交换机名称", "RoutingKey","对象",
message => {
message.getMessageProperties().setExpiration(String.valueOf(5000))
return message;
}
);
缺点:
该种方式可以创建一个承载不同超时时间消息的消息队列,但是这种方式有一个问题,如果消息队列中排在前面的消息没有到超时时间,即使后面的消息到了超时时间,先到超时时间的消息也不会进入死信队列,而是先检查排在最前面的消息队列是否到了超时时间,如果到了超时时间才会继续检查后面的消息。
第三种方式:使用rabbitmq的延时队列插件,实现同一个队列中有多个不同超时时间的消息,并按时间超时顺序出队
1、下载延迟插件
在 RabbitMQ 的 3.5.7 版本之后,提供了一个插件(rabbitmq-delayed-message-exchange)来实现延迟队列 ,同时需保证 Erlang/OPT 版本为 18.0 之后。
我这里 MQ 的版本是 3.10.0 现在去 GitHub 上根据版本号下载插件
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases
2、安装插件并启用
- 我用的是 Docker 客户端,下载完成后直接把插件放在 /opt/rabbitmq 目录,然后拷贝到容器内plugins目录下(rabbitmq是容器的name,也可以使用容器id)
docker cp /opt/rabbitmq/rabbitmq_delayed_message_exchange-3.10.0.ez rabbitmq:/plugins
- 进入 Docker 容器
docker exec -it rabbitmq /bin/bash
- 在plugins内启用插件
#先执行,解除防火墙限制,增加文件权限
cd plugins
umask 0022
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
- 退出容器
exit
- 重启 RabbitMQ
docker restart rabbitmq
- 通过UI查看
原理
代码使用
消费者
/*
* Copyright (c) 2020, 2024, fpl1116.cn All rights reserved.
*
*/
package com.fpl.consumers;
import cn.hutool.core.map.MapUtil;
import com.fpl.model.OrderingOk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Map;
/**
* <p>Project: spring-rabbitmq - DelayConsumer</p>
* <p>Powered by fpl1116 On 2024-04-09 11:34:07</p>
* <p>描述:<p>
*
* @author penglei
* @version 1.0
* @since 1.8
*/
@Configuration
@Slf4j
public class DelayConsumer {
@Bean
public Queue delayQueue1(){
return QueueBuilder.durable("Delay_Q01").lazy().build();
}
@Bean
public CustomExchange delayExchange(){
//参数x-delayed-type
Map<String, Object> map = MapUtil.of("x-delayed-type","direct");
return new CustomExchange("Delay_E01","x-delayed-message",true,false,map);
}
@Bean
public Binding binding1(Queue delayQueue1, CustomExchange delayExchange){
return BindingBuilder.bind(delayQueue1).to(delayExchange).with("RK01").noargs();
}
// @RabbitListener(queues = "Delay_Q01")
public void receiveMessage(OrderingOk msg) {
log.info("消费者1 收到消息:"+ msg );
}
}
生产者
/*
* Copyright (c) 2020, 2024, fpl1116.cn All rights reserved.
*
*/
package com.fpl.provider;
import com.fpl.model.OrderingOk;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* <p>Project: spring-rabbitmq - DelayProvider</p>
* <p>Powered by fpl1116 On 2024-04-09 11:35:51</p>
* <p>描述:<p>
*
* @author penglei
* @version 1.0
* @since 1.8
*/
@Service
public class DelayProvider {
@Autowired
private RabbitTemplate rabbitTemplate;
public void send(OrderingOk orderingOk) {
rabbitTemplate.convertAndSend("Delay_E01", "RK01", orderingOk,new MessagePostProcessor(){
@Override
public Message postProcessMessage(Message message) throws AmqpException {
int id = orderingOk.getId();
int ttl = 0;
if(id == 1){
ttl = 50*1000;
}else if(id == 2){
ttl = 30*1000;
}else if(id ==3){
ttl = 40*1000;
}else if(id ==4){
ttl = 10*1000;
}else if(id ==5){
ttl = 20*1000;
}
//延迟交换机使用的delay参数,设置消息的延期时长,单位是微妙
message.getMessageProperties().setDelay(ttl);
//延迟交换机消息默认是持久化的
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
return message;
}
});
}
}
测试
@Test
void test5() throws IOException {
for (int i = 1; i <=5;i++){
OrderingOk orderingOk = OrderingOk.builder().id(i).name("张 " + i).build();
delayProvider.send(orderingOk);
System.out.println("发送成功:"+i);
}
System.in.read();
}