如何通过RabbitMq实现动态定时任务详解

目录
  • 一、需求背景
  • 二、方案思考
    • (1)需求大致分析
    • (2)可尝试的方案
  • 三、通过RabbitMQ实现延时任务并间接实现动态定时任务。
    • (1)通过死信的方式实现延时信息消费
    • (2)通过MQ延时插件实现延时任务(重点)
  • 四、MQ延时任务插件实现动态定时任务
    • (1)安装延时插件
    • (2)编码实现
    • (3)流程图
  • 总结

一、需求背景

定时任务的需求所谓是数不胜数,其中实现方式也是百花齐放,用得最多的大概率为Springboot中的 @Scheduled(cron = “0 0 1 1 * ?”) 注解,或者是定时任务XXL-JOB框架,这两者我接触的比较多,除此之外还有,Quartz 、elastic-job、但这两个在分布式领域而言,和XXL-JOBB比较,XXL-JOB更为受欢迎。无论是这些框架或者是springboot自带的定时任务组件,基本上都能满足固定定时任务的需求。而我们今天讨论的是动态定时任务的实现。

动态定时任务的需求其实在现实生活中随处可见,如花费到期多久之后发送信息提醒用户?时间间隔是多少。又或者客户下单之后多久提醒商家发货,提醒的频率又是多少…。这样的需求还有很多。今天我们针对此类需求进行探讨。

二、方案思考

(1)需求大致分析

对于此类需求相比于传统的定时任务无非多了可控性, 其可控性包括了定时任务开始和结束时间的可控性,周期性可控性,只要解决了这两个问题,实际上此类的需求也就迎刃而解了。

(2)可尝试的方案

前面提供的方案只做文字探索性描述。

2.1、 采用重写Springboot 的定时框架,从数据库中读取cron表达式来实现可控性周期。

其本质是通过如下线程进行动态定时任务的创建,从而实现对应的周期可控性。

ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();

其具体的细节不再说,其存在的痛点包括了

1 . 需要另外逻辑去实现可控性开始时间和结束时间

2. 此任务开启的入参是corn表达式,需要另外的逻辑将其进行转化,太过于猥琐

2.2、采用时间线程池

时间线程池我忘记叫什么,他是可以指定开始时间,周期时间的,相对而言,比第一种方案来得更为直观,其我考虑到的痛点如下。其实上面那种方案也是有这个痛点的。

  1. 多节点,多服务的服务部署情况下,无法实现高可用特性
  2. 需要编写过多的逻辑来管理任务线程,如果不够谨慎,有可能造成内存浪费。

2.3、采用延时操作

简单言之,实际上只要实现了延时操作 便是实现了动态的开始时间以及周期性运行,可以利用其递归的概念实现所谓的动态周期。

redis 队列来实现延时

redis的体量本身定位就不高,在数据量(任务量)过大时,对redis的压力也很大,redis不一定扛得住。但其实通过redis来实现延时消息这样的成功案列还是有很多的。在这里就不细说了。

RabbitMq实现延时消息。

通过MQ实现延时消息是本文的重点,在标题三会细说。

三、通过RabbitMQ实现延时任务并间接实现动态定时任务。

(1)通过死信的方式实现延时信息消费

通过创建死信队列来实现延时任务,然后再通过递归思想实现对应的逻辑,就可以实现对应的动态延时任务,但是这个会存在以下下几个痛点。

队列顺序消费

通过死信,我们确实可以动态的控制消息的消费时间,但是消息在队列里面,如果队列里面存在多个信息任务,前一个未到消费时间,后一个已经到了消费时间,这就好导致了,即使后面任务信息消费时间到了,却没法被消费的问题。解决方法,对队列进行排序逻辑,但如果这样做的话,就有点猥琐了。

开销过大。

对于通过死信来实现延时消息,网上有挺多优秀的博客介绍,在此就不做说明了。

(2)通过MQ延时插件实现延时任务(重点)

使用延时插件需要MQ在3.6以上(实际上我在尝试下载的时候并未发现git上有对应3.6的插件,所以还是选择较高的版本比较好)。

四、MQ延时任务插件实现动态定时任务

(1)安装延时插件

这里不做过多说明,重点在于编码的实现,主要步骤如下。

去官网下载对应版本的插件,地址为下载地址

插件名字为rabbitmq_delayed_message_exchange

将插件放到MQ插件目录下,然后cmd命令解压网(网上有命令),然后重启mq服务。大概就这样的一个过程。

(2)编码实现

创建队列

这里只弄了对应的核心代码,大致就是创建延时交换机,延时队列,以及绑定器,对应的key,value如下

    public static final String DELAY_EXCHANGE = "delay.exchange";

    public static final String DELAY_ROUTE_KEY = "delay.routeKey";

    public static final String DELAY_QUEUE = "delay.queue";

    /**
     * 延时交换机
     * @return 延时交换机
     */
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> arguments = new HashMap<>(1);
        arguments.put("x-delayed-type", "direct");
        return new CustomExchange(DELAY_EXCHANGE,"x-delayed-message",true,false,arguments);
    }

    /**
     * mq已经安装了延时插件使用,否则得使用延时插件
     * @return 延时发送队列。
     */
    @Bean
    public Queue delayQueue() {
        return new Queue(DELAY_QUEUE,true,false,false);
    }

    /**
     * 延时绑定区
     * @return 延时绑定区
     */
    @Bean
    public Binding delayBind() {
        return BindingBuilder.bind(this.delayQueue()).to(this.delayExchange()).with(DELAY_ROUTE_KEY).noargs();
    }

生产者

这里写得比较随意,也直接使用了lombok,还直接用了 @Service ,有点草率,主要为了让读者看得清晰。还用了hutool工具类的JSONUtil。

可以清晰的看到主方法里面需要传一个Integer类型的入参,这个时间我将其转换成了秒,其MQ实际入参为毫秒,所以读者不要被误导。入参time通俗的讲就是这个消息多久之后被消费。不需要在乎顺序。

package com.linkyoyo.bill.mq.impl;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.linkyoyo.bill.bo.WorkOrderDelaySenMailActionBO;
import com.linkyoyo.bill.config.RabbitMQConfig;
import com.linkyoyo.bill.mq.DelaySenderService;
import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

/**
 * 延时发送
 * @author 邹 [295006967@qq.com]
 * @date 2022/1/4 20:33
 */
@Slf4j
@RequiredArgsConstructor
@AllArgsConstructor
@Service
public class DelaySenderServiceImpl implements DelaySenderService {

    private final RabbitTemplate rabbitTemplate;

    @Override
    @Async
    public void sendMessageByDelay(JSONObject message, Integer time) {
        if(ObjectUtil.isNull(message) || ObjectUtil.isNull(time)) {
            return;
        }
        rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_EXCHANGE, RabbitMQConfig.DELAY_ROUTE_KEY, message, msg -> {
            msg.getMessageProperties().setHeader("x-delay", time * 1000);
            return msg;
        });
        log.info("延时发送成功:延时周期时间{}毫秒,消息内容为{}", time * 1000, message);
    }

    @Override
    public void sendMessageByDelay(WorkOrderDelaySenMailActionBO actionBO) {
        Integer afterSecond = actionBO.getAfterSecond();
        if(ObjectUtil.isNull(afterSecond)) {
            afterSecond = 0;
        }
        this.sendMessageByDelay(JSONUtil.parseObj(actionBO), afterSecond);
    }
}

消费者

消费者的demo不太好写,只是做了一个简单的伪代码。 以定时任务发邮箱为例

1- 消费者线程开始,先执行发邮箱任务

2- 发送完邮箱之后判断是否还需要发邮箱,如果需要,就再通过生产者发送延时邮箱 此时可以指定下一次消费的时间,以此流程走下去便是一套动态任务的流程实现。可以参考后续的流程图。

这样就实现一个简易的定时任务发送邮箱的逻辑

	private final DelaySenderService delaySenderService;

    @RabbitHandler
    @RabbitListener(queues = RabbitMQConfig.DELAY_QUEUE)
    public void delayConsumer(Message message) {
        //业务逻辑
        this.sendMail(workOrderDelaySenMailActionBO);
        // 判断是否需要递归执行定时任务(实际上就是使用生产者再发一次延时消息,确认下一次消费)
        if(需要进行定时任务) {
             this.sendDelayMessageToMq(workOrderDelaySenMailActionBO);
        }
        log.info("信息为:{}", message.getBody());
    }

大致流程就这么多了,以下是整套步骤流闭环程图

(3)流程图

总结

到此这篇关于如何通过RabbitMq实现动态定时任务的文章就介绍到这了,更多相关RabbitMq实现动态定时任务内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

时间: 2022-01-12

SpringBoot下RabbitMq实现定时任务

本文实例为大家分享了SpringBoot下RabbitMq实现定时任务,供大家参考,具体内容如下 定时任务场景:订单下单15分钟未付款自动关闭 延迟任务实现原理图如下: 根据上图看出我们需要两个队列(一是死信队列,消息在里面度过TLL时间,二是处理队列,消息度过TLL时间后进入该队列),两个交换机和路由(一是用来将消息送入死信队列,二是将消息从死信队列送到处理队列),但是交换机其实可以用同一个,也就是一个交换机搭配两个路由的方式. 以下为代码实现过程: //首先rabbitAdmin的配置 @B

Rabbitmq延迟队列实现定时任务的方法

场景 开发中经常需要用到定时任务,对于商城来说,定时任务尤其多,比如优惠券定时过期.订单定时关闭.微信支付2小时未支付关闭订单等等,都需要用到定时任务,但是定时任务本身有一个问题,一般来说我们都是通过定时轮询查询数据库来判断是否有任务需要执行,也就是说不管怎么样,我们需要先查询数据库,而且有些任务对时间准确要求比较高的,需要每秒查询一次,对于系统小倒是无所谓,如果系统本身就大而且数据也多的情况下,这就不大现实了,所以需要其他方式的,当然实现的方式有多种多样的,比如Redis实现定时队列.基于优先

详解Spring Cloud Stream使用延迟消息实现定时任务(RabbitMQ)

我们在使用一些开源调度系统(比如:elastic-job等)的时候,对于任务的执行时间通常都是有规律性的,可能是每隔半小时执行一次,或者每天凌晨一点执行一次.然而实际业务中还存在另外一种定时任务,它可能需要一些触发条件才开始定时,比如:编写博文时候,设置2小时之后发送.对于这些开始时间不确定的定时任务,我们也可以通过Spring Cloud Stream来很好的处理. 为了实现开始时间不确定的定时任务触发,我们将引入延迟消息的使用.RabbitMQ中提供了关于延迟消息的插件,所以本文就来具体介绍

详解spring cloud config实现datasource的热部署

关于spring cloud config的基本使用,前面的博客中已经说过了,如果不了解的话,请先看以前的博客 spring cloud config整合gitlab搭建分布式的配置中心 spring cloud config分布式配置中心的高可用 今天,我们的重点是如何实现数据源的热部署. 1.在客户端配置数据源 @RefreshScope @Configuration// 配置数据源 public class DataSourceConfigure { @Bean @RefreshScope

详解spring cloud config整合gitlab搭建分布式的配置中心

在前面的博客中,我们都是将配置文件放在各自的服务中,但是这样做有一个缺点,一旦配置修改了,那么我们就必须停机,然后修改配置文件后再进行上线,服务少的话,这样做还无可厚非,但是如果是成百上千的服务了,这个时候,就需要用到分布式的配置管理了.而spring cloud config正是用来解决这个问题而生的.下面就结合gitlab来实现分布式配置中心的搭建.spring cloud config配置中心由server端和client端组成, 前提:在gitlab中的工程下新建一个配置文件config

详解spring cloud构建微服务架构的网关(API GateWay)

前言 在我们前面的博客中讲到,当服务A需要调用服务B的时候,只需要从Eureka中获取B服务的注册实例,然后使用Feign来调用B的服务,使用Ribbon来实现负载均衡,但是,当我们同时向客户端暴漏多个服务的时候,客户端怎么调用我们暴漏的服务了,如果我们还想加入安全认证,权限控制,过滤器以及动态路由等特性了,那么就需要使用Zuul来实现API GateWay了,下面,我们先来看下Zuul怎么使用. 一.加入Zuul的依赖 <dependency> <groupId>org.spri

详解spring cloud使用Hystrix实现单个方法的fallback

本文介绍了spring cloud-使用Hystrix实现单个方法的fallback,分享给大家,具体如下: 一.加入Hystrix依赖 <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-hystrix</artifactId> </dependency> 二.编写Controller package c

详解Spring Cloud 跨服务数据聚合框架

AG-Merge Spring Cloud 跨服务数据聚合框架 解决问题 解决Spring Cloud服务拆分后分页数据的属性或单个对象的属性拆分之痛, 支持对静态数据属性(数据字典).动态主键数据进行自动注入和转化, 其中聚合的静态数据会进行 一级混存 (guava). 举个栗子: 两个服务,A服务的某张表用到了B服务的某张表的值,我们在对A服务那张表查询的时候,把B服务某张表的值聚合在A服务的那次查询过程中 示例 具体示例代码可以看 ace-merge-demo 模块 |------- ac

详解spring cloud整合Swagger2构建RESTful服务的APIs

前言 在前面的博客中,我们将服务注册到了Eureka上,可以从Eureka的UI界面中,看到有哪些服务已经注册到了Eureka Server上,但是,如果我们想查看当前服务提供了哪些RESTful接口方法的话,就无从获取了,传统的方法是梳理一篇服务的接口文档来供开发人员之间来进行交流,这种情况下,很多时候,会造成文档和代码的不一致性,比如说代码改了,但是接口文档没有改等问题,而Swagger2则给我们提供了一套完美的解决方案,下面,我们来看看Swagger2是如何来解决问题的. 一.引入Swag

详解Spring Cloud Feign 熔断配置的一些小坑

1.在使用feign做服务调用时,使用继承的方式调用服务,加入Hystrix的熔断处理fallback配置时,会报错,已解决. 2.使用feign默认配置,熔断不生效,已解决. 最近在做微服务的学习,发现在使用feign做服务调用时,使用继承的方式调用服务,加入Hystrix的熔断处理fallback配置时,会报错,代码如下: @RequestMapping("/demo/api") public interface HelloApi { @GetMapping("user/

详解Spring cloud使用Ribbon进行Restful请求

写在前面 本文由markdown格式写成,为本人第一次这么写,排版可能会有点乱,还望各位海涵.  主要写的是使用Ribbon进行Restful请求,测试各个方法的使用,代码冗余较高,比较适合初学者,介意轻喷谢谢. 前提 一个可用的Eureka注册中心(文中以之前博客中双节点注册中心,不重要) 一个连接到这个注册中心的服务提供者 一个ribbon的消费者 注意:文中使用@GetMapping.@PostMapping.@PutMapping.@DeleteMapping等注解需要升级 spring

详解Spring Cloud微服务架构下的WebSocket解决方案

WebSocket在现代浏览器中的应用已经算是比较普遍了,在某些业务场景下,要求必须能够在服务器端推送消息至客户端.在没有WebSocket的年代,我们使用过dwr,在那个时候dwr真实一个非常棒的方案.但是在WebSocket兴起之后,我们更愿意使用标准实现来解决问题. 首先交代一下,本篇文章不讲解WebSocket的配置,主要讲的是针对在微服务架构集群模式下解决方案的选择. 微服务架构大家应该都不陌生了,在微服务架构下,服务是分布式的,而且为了保证业务的可用性,每个服务都是以集群的形式存在.