当前位置：首页 > news >正文

RabbitMQ实现延迟消息发送——实战篇

news 来源：原创 2025/6/15 17:25:07

在项目中，我们经常需要使用消息队列来实现延迟任务，本篇文章就向各位介绍使用RabbitMQ如何实现延迟消息发送，由于是实战篇，所以不会讲太多理论的知识，还不太理解的可以先看看MQ的延迟消息的一个实现原理再来看这篇文章跟着练手哦~

需求背景

我这儿的一个需求背景大概是干部添加完活动后，由管理员进行审批，审批通过后，该活动id连同设置的过期时间会被放入消息队列中，等到活动结束时间到的时候，自动将活动的状态设置为已完成，这里华丽一个活动图，各位参考一下。

需求了解完之后我们就可以开始的写代码啦~（手动微笑）

代码部分

首先，我们需要定义一个死信交换机和死信队列，用来接收来自普通队列的消息。

//    创建死信交换机，处理延迟消息通知@Bean("dead_letter_exchange")public DirectExchange delayExchange(){return new DirectExchange("dead_letter_exchange",true,false);}

//    创建死信队列public Queue deadLetterQueue(){Queue queue = new Queue("dead_letter_queue", true);rabbitAdmin.declareQueue(queue);log.info("死信队列声明成功：" + queue.getName());return queue;    }

然后，我们需要配置一个普通的消息队列和一个普通的交换机，这个消息队列需要设置对应的死信交换机和死信路由，同时我们这个普通队列需要接收一个过期时间，保证一到过期时间消息就会被发送到死信队列当中。

//    创建一个普通队列,接受一个过期时间，出列活动结束后，发送到死信队列public Queue normalQueue(Long expireTime){Map<String,Object> args = new HashMap<>();if (expireTime != null && expireTime > 0) {  // 确保 TTL 是正数args.put("x-message-ttl", expireTime);}// 设置死信交换机args.put("x-dead-letter-exchange",deadLetterExchange);// 设置死信路由键args.put("x-dead-letter-routing-key","dead_letter_routing_key");Queue queue = new Queue("normal_queue", true, false, false, args);log.info("普通队列声明成功：" + queue.getName());return queue;    }

//    创建一个普通交换机，处理活动结束自动设置活动状态为结束@Bean("activity_end_exchange")public DirectExchange activityEndExchange(){return new DirectExchange("activity_end_exchange");}

然后我们需要分别将死信交换机和死信队列，普通交换机和普通队列分别进行绑定。

//    将死信队列和死信交换机进行绑定public void bindDeadLetterRouting(){Queue queue=queueDeclareConfig.deadLetterQueue();Binding binding = BindingBuilder.bind(queue).to(deadLetterExchange).with("dead_letter_routing_key");rabbitAdmin.declareBinding(binding);log.info("死信队列绑定成功，死信队列名称----》" + queue.getName() + "，死信交换机名称----》" + deadLetterExchange.getName());}//    绑定活动结束交换机和普通队列public void bindActivityEndRouting(Long expireTime) {Queue queue = queueDeclareConfig.normalQueue(expireTime);Binding binding = BindingBuilder.bind(queue).to(activityEndExchange).with("activity_end_routing_key");rabbitAdmin.declareBinding(binding);}

当然，我们还需要配置生产者来发送消息到交换机里面

//活动结束后，发送消息到死信队列，自动设置活动结束状态public void sendActivityEndMessage(Long expireTime, Integer activityId) {rabbitMQBindRoutingConfig.bindDeadLetterRouting();rabbitMQBindRoutingConfig.bindActivityEndRouting(expireTime);try {// 将消息发送到普通队列，等待消息过期发送到死信交换机rabbitTemplate.convertAndSend("activity_end_exchange", "activity_end_routing_key", activityId, msg -> {msg.getMessageProperties().setExpiration(expireTime.toString());return msg;});} catch (Exception e) {log.error("发送消息失败------->" + activityId);throw new RuntimeException("发送消息失败---->" + activityId);}}

这里生产者的代码可以根据你的业务逻辑具体进行更改~

消费者逻辑也需要进行编写一下

//    使用MQ延迟队列，活动结束，修改活动状态@RabbitListener(queues = "dead_letter_queue")public void updatePlaceOccupyStatus(Message message, Channel channel){try {String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);Integer activityId = Integer.parseInt(messageBody);ActivityInfo activityInfo = baseMapper.selectById(activityId);LambdaUpdateWrapper<ActivityInfo> wrapper = new LambdaUpdateWrapper<>();wrapper.eq(ActivityInfo::getActivityId,activityId).set(ActivityInfo::getProgress,StatusConstant.FINISH);if(baseMapper.update(activityInfo,wrapper)>0){channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}} catch (Exception e) {log.error("处理消息时发生错误：" + e.getMessage());try {channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);} catch (IOException ioException) {ioException.printStackTrace();}}

消费者这边需要注意的是如果你选择的提交类型不是自动提交的话，在处理完消息之后需要手动ack一下消息，不然消费的消息不会被认为已经消费，从而导致消息积压，也会在之后的消费中重复进行消费，因此你需要告诉生产者这条消息已经被消费了。

当然，如果在消费的过程中出现了什么问题，可以设置以下这行代码：

basicNack方法接收三个参数：
deliveryTag: 消息的标识符。
multiple: 是否对多个消息进行否定确认。
requeue: 是否将消息重新放入队列。

可以根据你的需求进行设定~

然后的然后，我们需要再application.yml当中进行配置相关信息：

rabbitmq:host: localhostport: 5672username: guestpassword: guest
#    确认消息发送到交换机上publisher-confirm-type: correlated#    消息发送到队列确认,失败回调publisher-returns: truelistener:direct:acknowledge-mode: manualretry:enabled: true
#          重试的时间间隔为1sinitial-interval: 1000ms
#          最大重试3次max-attempts: 3
#          最大的重试时间间隔为2smax-interval: 2000ms
#          每次重试时间间隔为1s，每次重试时间间隔倍数multiplier: 1.0#重试次数超过上面的设置之后是否丢弃（false不丢弃时需要写相应代码将该消息加入死信队列）default-requeue-rejected: falsesimple:default-requeue-rejected: falseacknowledge-mode: manual
#        最小消费者数量concurrency: 1
#        最大消费者数量max-concurrency: 10retry:enabled: trueinitial-interval: 1000msmax-attempts: 3max-interval: 2000msmultiplier: 1.0

上面给出了一个比较全的配置，你可以根据你的需求进行选择，但是需要注意的是default-requeue-rejected: false这一行配置一定要先配置，不然你的消息在普通队列中过期了，是不会发送到死信队列当中进行消费的~

到这儿，基本上所有的代码都写的差不多了，当然我们还需要再rabbitmq控制平台上分别建一个普通交换机和一个死信交换机，一个普通队列和一个私信队列，然后分别绑定就可以了。

注意的是，普通交换机也需要在平台上配置一次死信队列和死信路由：

到这儿，如果没有什么问题的话基本上已经可以直接运行了，所以我的这篇文章到这儿基本上也已经结束了，如果你有什么问题，可以评论区留言，我们相互学习~

需求背景

相关知识点拓展

1. Direct Exchange（直连交换机）

2. Fanout Exchange（扇出交换机）

3. Topic Exchange（主题交换机）

4. Headers Exchange（头交换机）

5. Default Exchange（默认交换机）

死信交换机适合使用哪种类型？

推荐选择

代码部分

相关文章：