RabbitMQ消息队列怎么实现延迟任务

发布时间:2022-04-06 14:00:53 作者:iii
来源:亿速云 阅读:178

这篇文章主要介绍“RabbitMQ消息队列怎么实现延迟任务”的相关知识,小编通过实际案例向大家展示操作过程,操作方法简单快捷,实用性强,希望这篇“RabbitMQ消息队列怎么实现延迟任务”文章能帮助大家解决问题。

一、序言

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

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

1、实现原理

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

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

2、组件选型

RabbitMQ消息队列怎么实现延迟任务

二、方案设计

(一)服务器

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

(二)生产者

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

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

RabbitMQ消息队列怎么实现延迟任务

(三)消费者

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

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

三、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>

关于“RabbitMQ消息队列怎么实现延迟任务”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识,可以关注亿速云行业资讯频道,小编每天都会为大家更新不同的知识点。

推荐阅读:
  1. RabbitMQ如何实现RPC远程调用消息队列
  2. SpringBoot:初探 RabbitMQ 消息队列

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

rabbitmq

上一篇:Vue+ Antv F2怎么实现层叠柱状图

下一篇:Java数组高频考点实例分析

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》