乐趣区

关于消息队列:基于消息队列RabbitMQ实现延迟任务

jiezi

一、序言

提早工作利用宽泛，提早工作典型利用场景有 订单超时主动勾销 ； 领取回调重试。其中订单超时勾销具备幂等性属性，无需思考反复生产问题；领取回调重试须要思考反复生产问题。

提早工作具备如下特点：在将来的某个工夫点执行；个别仅执行一次。

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>

喜爱本文点个♥️赞♥️反对一下，如有须要，可通过微信 dream4s 与我分割。相干源码在 GitHub，视频解说在 B 站，本文珍藏在博客天地。

退出移动版