SpringBoot Rabbitmq3.9 DLX方式实现延迟队列

MQ安装教程

https://blog.csdn.net/L630642270/article/details/126767808

安装RabbitMQ的延迟插件

1.下载：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.9.0/rabbitmq_delayed_message_exchange-3.9.0.ez
2.将插件文件复制到RabbitMQ安装目录的plugins目录下:
cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.9.11/plugins
3.进入RabbitMQ安装目录的sbin目录下，使用如下命令启用延迟插件
// 进入目录
cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.9.11/sbin
// 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

Maven依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置文件

# 应用名称
spring.application.name=rabbitMq
# 应用服务 WEB 访问端口
server.port=8080
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
# 开启消息确认机制 confirm 异步
spring.rabbitmq.publisher-confirm-type=correlated
# 之前的旧版本 开启消息确认机制的方式
# spring.rabbitmq.publisher-confirms=true
# 开启return机制
spring.rabbitmq.publisher-returns=true
# 消息开启手动确认
spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 消费者每次从队列获取的消息数量。此属性当不设置时为：轮询分发，设置为1为：公平分发
spring.rabbitmq.listener.simple.prefetch=1
# 是否支持重试
spring.rabbitmq.listener.simple.retry.enabled=true#消费者最小数量
spring.rabbitmq.listener.simple.concurrency=1
#消费之最大数量
spring.rabbitmq.listener.simple.max-concurrency=10

常量池

/*** @AUTHOR Summer* @DATE 2022-06-28 15:19*/
public interface Consts {// 延迟交换机String COURSE_TASK_DELAY_EXCHANGE = "course.task.delayExchange";// 延迟队列String COURSE_TASK_DELAY_QUEUE = "course.task.delayQueue";
}

MQ配置

/*** @AUTHOR Summer* @DATE 2022-08-23 10:22*/
@Configuration
public class RabbitmqConfig {@Beanpublic CustomExchange delayExchange() {Map<String, Object> args = new HashMap<>(1);args.put("x-delayed-type", "direct");return new CustomExchange(Consts.COURSE_TASK_DELAY_EXCHANGE, "x-delayed-message", true, false, args);}/*** 延时队列*  DLX方式* @return*/@Beanpublic Queue immediateQueue() {// 第一个参数是创建的queue的名字，第二个参数是是否支持持久化return new Queue(Consts.COURSE_TASK_DELAY_QUEUE, true);}/*** 给延时队列绑定交换机*   DLX方式* @return*/@Beanpublic Binding bindingNotify() {return BindingBuilder.bind(immediateQueue()).to(delayExchange()).with(Consts.COURSE_TASK_DELAY_EXCHANGE).noargs();}
}

生产者

@Autowired
private RabbitTemplate rabbitTemplate;public void sendMqMessage(){User user = new User(1,"张三");// rabbitTemplate.setMandatory(true);// rabbitTemplate.setConfirmCallback(confirmCallback);CorrelationData correlationData = new CorrelationData("confirm-" + System.currentTimeMillis());rabbitTemplate.convertAndSend(Consts.COURSE_TASK_DELAY_EXCHANGE,Consts.COURSE_TASK_DELAY_EXCHANGE,user,message -> {// 根据业务修改时间Integer taskDuration = 5000; // 毫秒message.getMessageProperties().setDelay(taskDuration);  // 毫秒return message;},correlationData);
}

消费者

/*** @AUTHOR Summer* @DATE 2022-08-19 15:48*/
@Slf4j
@Component
public class CourseComsumer {@Autowiredprivate CourseActivityUserService courseActivityUserService;/*** 延迟队列-消费方* @return* @author Summer* @date 2022-08-23 11:58*/@RabbitListener(queues = Consts.COURSE_TASK_DELAY_QUEUE)public void getDLXMessage(User user, Channel channel, Message message) throws IOException {try {channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);log.error("DLX延迟队列消息:{}",user);} catch (IOException e) {if (message.getMessageProperties().getRedelivered()) {log.error("DLX延迟队列消息已重复处理失败,拒绝再次接收...");// 拒绝消息//basicReject：拒绝消息，与basicNack区别在于不能进行批量操作，其他用法很相似。//deliveryTag：表示消息投递序号。//requeue：值为 true 消息将重新入队列。channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);} else {log.error("DLX延迟队列消息即将再次返回队列处理...");//basicNack ：表示失败确认，一般在消费消息业务异常时用到此方法，可以将消息重新投递入队列。//deliveryTag：表示消息投递序号。//multiple：是否批量确认。//requeue：值为 true 消息将重新入队列。channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}}}
}

SpringBoot Rabbitmq3.9 DLX方式实现延迟队列相关推荐

RabbitMQ实现延迟队列的方式
1.背景最近在做类似拍卖系统的上架功能,卖家上架物品以后,例如到期时间24小时或者48小时,如果无竞拍者或者购买者,则物品自动下架到用户的邮件中.诸如电商用户下单,30分钟未支付,则自动取消订单,归 ...
RabbitMQ如何实现延迟队列？
欢迎支持笔者新作:<深入理解Kafka:核心设计与实践原理>和<RabbitMQ实战指南>,同时欢迎关注笔者的微信公众号:朱小厮的博客. 欢迎跳转到本文的原文链接:https: ...
完整案例：实现延迟队列的两种方法
延迟队列是指把当前要做的事情,往后推迟一段时间再做. 延迟队列在实际工作中和面试中都比较常见,它的实现方式有很多种,然而每种实现方式也都有它的优缺点,接下来我们来看. 延迟队列的使用场景延迟队列的常 ...
你真的知道怎么实现一个延迟队列吗？
原文地址:https://mp.weixin.qq.com/s/jL8_23pjYWV74rsjoWNPWg 目录前言延迟队列定义应用场景实现方案 Redis zset TimeWheel 时 ...
SpringBoot之使用RabbitMQ实现延迟队列
在我们的各个项目中,经常会有这样的需求. 订单模块:在订单下单后30分钟如果没有付款,就自动取消订单, 短信模块:在下单成功后60s给用户发送短信通知支付模块:在微信/支付宝支付成功后,1分钟后去调 ...
RabbitMQ,Springboot整合RabbitMQ实现消息可靠性投递,Consumer ACK,TTL,死信队列,使用TTL+死信队列=延迟队列
搭建SpringBoot项目,用于演示 springboot版本 <dependency><groupId>org.spri ...
SpringBoot整合RabbitMQ 消息可靠投递、手动ack、延迟队列、死信队列、消息幂等性保障、消息积压
1.消息可靠投递在使用 RabbitMQ 的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败场景.RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式. confirm 确认模式 ...
SpringBoot RabbitMQ 延迟队列代码实现
场景用户下单后,如果30min未支付,则删除该订单,这时候就要可以用延迟队列准备利用rabbitmq_delayed_message_exchange插件: 首先下载该插件:https://ww ...
SpringBoot RabbitMQ 集成七延迟队列
为什么80%的码农都做不了架构师?>>> 何为延迟队列? 顾名思义,延迟队列就是进入该队列的消息会被延迟消费的队列.而一般的队列,消息一旦入队了之后就会被消费者马上消费. 延迟 ...

SpringBoot Rabbitmq3.9 DLX方式实现延迟队列

MQ安装教程

安装RabbitMQ的延迟插件

Maven依赖

配置文件

常量池

MQ配置

生产者

消费者

SpringBoot Rabbitmq3.9 DLX方式实现延迟队列相关推荐

最新文章

热门文章