RabbitMQ实现延迟消费(延迟队列)

Posted 2023-04-18

参考技术A

首先在pom中加入依赖

然后配置yml文件

创建业务队列与死信队列

该消费者是消费死信队列中的消息

启动服务之后，可以看到创建的交换机和队列

消息过期之后从 prod_queue_pay 队列转发到 dl-queue 队列。很好的实现了消息延迟消费。但我们会发现一个问题，通过给队列属性设置过期时间，如果我现在有不同的场景，比如我5s、10s、15s之后延迟消费，那需要创建三个队列。每次有一个不同的时间段的需求过来，我都需要创建一个队列，这肯定不行。

快速入口： https://www.jianshu.com/p/78354a3e35d0

发送消息的时候通过在header添加"x-delay"参数来控制消息的延时时间

启动服务，登录RabbitMQ管理界面，可以看到交换机和队列都已经创建成功。

该类型消息支持延迟投递机制,接收到消息后并未立即将消息投递至目标队列中，而是存储在mnesia(一个分布式数据系统)表中，检测消息延迟时间，如达到可投递时间时并将其通过x-delayed-type类型标记的交换机类型投递至目标队列。这些判断和操作导致效率不如普通的Exchange，所以如果不需要的话，就不要用插件类型的延迟队列。

基于消息队列（RabbitMQ）实现延迟任务

一、序言

延迟任务应用广泛，延迟任务典型应用场景有订单超时自动取消；支付回调重试。其中订单超时取消具有幂等性属性，无需考虑重复消费问题；支付回调重试需要考虑重复消费问题。

延迟任务具有如下特点：在未来的某个时间点执行；一般仅执行一次。

1、实现原理

生产者将带有延迟信息的消息发送到RabbitMQ交换机中，等待延迟时间结束方将消息转发到绑定的队列中，消费者通过监听队列消费消息。延迟任务的关键在消息在交换机中停留。

显而易见，基于RabbitMQ实现延迟任务对服务器的可靠性要求极高，交换机内部消息无持久化机制，比如单机模式服务重启，未开始的延迟任务均丢失。

2、组件选型

二、方案设计

（一）服务器

RabbitMQ服务需要安装x-delayed-message插件以处理延迟消息。

（二）生产者

延迟任务的实现对生产者的要求是将消息可靠的投递到交换机，因此使用confirm确认机制即可。

订单生成之后，先入库，然后以订单ID为key将订单详情存入Redis中（持久化），向RabbitMQ发送异步confirm确定请求。如果收到正常投递返回，则删除Redis中订单ID为key的数据，回收内存，否则以订单ID为key，从Redis中查询出订单数据，重新发送。

（三）消费者

延迟任务的实现对消费者的要求是以信息不丢失的方式消费消息，具体表现在：手动确认消息的消费，防止消息丢失；消费端持续稳定，防止消息堆积；消息消费失败有重试机制。

考虑到订单延迟取消属于幂等性操作，因此无需考虑消息的重复消费问题。

三、SpringBoot实现

实现部分仅贴一部分核心源码，完整项目请访问GitHub。

（一）生产者

考虑到下单是极为重要的操作，因此首先将订单落库、存盘，然后进行后续操作。

for (long i = 1; i <= 10; i++) 
    /* 1.模拟生成订单 */
    BuOrder order = createOrder(i);
    /* 2.订单入库 */
    orderService.removeById(order);
    orderService.saveOrUpdate(order);
    /* 3.将订单存入信息Redis */
    RedisUtils.setObject(RabbitTemplateConfig.ORDER_PREFIX + i, order);
    /* 4.向RabbitMQ异步投递消息 */
    rabbitTemplate.convertAndSend(RabbitmqConfig.DELAY_EXCHANGE_NAME, RabbitmqConfig.DELAY_KEY, order, RabbitUtils.setDelay(30000), RabbitUtils.correlationData(order.getOrderId()));

生产者可靠投递消息

public void confirm(CorrelationData correlationData, boolean ack, String cause) 
    if (correlationData == null) 
        return;
    
    String key = ORDER_PREFIX + correlationData.getId();
    if (ack) 
        /* 如果消息投递成功，则删除Redis中订单数据，回收内存 */
        RedisUtils.deleteObject(key);
     else 
        /* 从Redis中读取订单数据，重新投递 */
        BuOrder order = RedisUtils.getObject(key, BuOrder.class);
        /* 重新投递消息 */
        rabbitTemplate.convertAndSend(RabbitmqConfig.DELAY_EXCHANGE_NAME, RabbitmqConfig.DELAY_KEY, order, RabbitUtils.setDelay(30000), RabbitUtils.correlationData(order.getOrderId()));
    

（二）消费者

消费者端手动确认，避免消息丢失；失败自动重试。

@RabbitListener(queues = RabbitmqConfig.DELAY_QUEUE_NAME)
public void consumeNode01(Channel channel, Message message, BuOrder order) throws IOException 
    if (Objects.equals(0, order.getOrderStatus())) 
        /* 修改订单状态，设置为关闭状态 */
        orderService.updateById(new BuOrder(order.getOrderId(), -1));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info(String.format("消费者节点01消费编号为【%s】的消息", order.getOrderId()));
    

消费者可靠消费应至少开启两个及以上应用，确保消息队列中不积压消息。

（三）通用工具包

上述代码涉及一个工具类RabbitUtils，存在于如下依赖中，主要封装RabbitMQ极常用的工具方法。

<dependency>
  <groupId>xin.altitude.cms</groupId>
  <artifactId>ucode-cms-common</artifactId>
  <version>1.4.3.1</version>
</dependency>

愿与诸君共进步，大量的面试题及答案还有资深架构师录制的视频录像：有Spring，MyBatis，Netty源码分析，高并发、高性能、分布式、微服务架构的原理，JVM性能优化、分布式架构等这些成为架构师必备的知识体系
可以关注小编+转发文章+私信【架构资料】获取，最后祝大家都能拿到自己心仪的offer