分布式任务调度 xxl-job


1. 集中式任务调度

集中式任务与分布式任务调度恰好相反的概念,集中式任务就是单机任务,一个项目,一台服务器,也就是我们常说的单体应用。对于集中式任务,也就是我们Java开发中常见的定时任务。

1.1 定时任务

定时任务是指在指定时间去执行任务(业务代码);

定时任务基本概念图
在这里插入图片描述
定时任务应用场景
在这里插入图片描述

1.2 Java实现定时任务方式

(1) while(true) + Thread.sleep

(2) java.util.Timer + java.util.TimerTask
Timer是一种定时器工具,用来在一个后台线程计划执行指定任务,它可以计划执行一个任务一次或反复多次。

(3) ScheduledExecutorService
ScheduledExecutorService 是从jdk1.5开始为并发工具类被引入,是最理想的定时任务实现方式。

(4) Quartz
Quartz是一个开源的定时任务调度框架,由Java编写而成,用于Java生态下的定时任务调度,是一个灵活方便、使用简单的定时任务调度框架,可以和Spring整合使用。

(5) Spring Task
Spring框架提供的轻量级的定时任务调用工具,使用方便;

(6) SpringBoot注解@EnableScheduling + @Scheduled ; 底层依赖采用Spring Task;

1.3 集中式任务调度的缺陷

  • 不支持分片任务. 处理有序数据时, 多机器分片执行任务处理不同数据.
  • 不支持生命周期统一管理. 不重启服务情况下关闭, 启动任务.
  • 不支持集群, 存在任务重复执行的问题.
  • 不支持失败重试. 出现异常后任务终结, 不能根据执行状态控制任务重新执行.
  • 不支持动态调整. 不重启服务的情况下修改任务参数.
  • 无报警机制. 任务失败之后没有报警机制.
  • 任务数据难以统计. 任务数据量大时, 对于任务执行情况无法高效的统计执行情况.

2. 分布式任务调度

2.1 解决方案

由于集中式的定时任务调度需要解决一系列问题,所以在演进的过程中产生一些解决办法

  • 数据库唯一约束;
  • 使用配置文件、redis、mysql作为调度的开关;
  • 使用分布式锁实现调度的控制;
  • 使用分布式任务调度平台 TBSchedule、Elastic-Job、Saturn、XXL-JOB等;
  • 自研框架。

2.2 分布式任务调度的好处

解决集中式任务调度的先天缺陷.

  • 高可用. 集群架构下, 有节点出现异常, 不影响任务的执行.
  • 动态配置. 对任务执行周期, 及其他跟任务相关的属性不停机修改.
  • 生命周期管理. 可以不停机的情况下对任务单次执行启动任务, 停止任务进行管理.
  • 失败机制. 在任务执行过程中出现执行失败, 可以支持报警, 任务重试并快速查阅执行日志.
  • 数据统计. 定时任务数比较多的情况, 有多少任务, 哪些任务执行失败过, 哪些任务执行成功.
  • 分片执行. 对于批量处理的数据, 让对台机器执行该任务, 对处理数据进行分片处理.

2.3 开源框架

TBSchedule
淘宝推出的开源分布式任务调用系统,在很多互联网公司应用很广,但现在没用更新维护了;
GitHub地址: https://github.com/taobao/TBSchedule

Elastic-Job
当当网推出的分布式任务调度框架,很多公司的产品在使用该分布式任务调度框架;
官网 http://elasticjob.io

Saturn
唯品会推出的开源分布式任务调度平台,它是基于Elastic-Job开发的,新增了一些特性,在唯品会内部以及一些互联网公司在使用。
GitHub地址: https://github.com/vipshop/Saturn

PowerJob
新一代分布式任务调度与计算框架,支持CRON、API、固定频率、固定延迟等调度策略,提供工作流来编排任务解决依赖关系,使用简单,功能强大,文档齐全。
官网: http://www.powerjob.tech/
GitHub开源地址: https://github.com/KFCFans/PowerJob
Gitee开源地址: https://gitee.com/KFCFans/PowerJob

XXL-JOB
美团开源的轻量级分布式任务调度平台,其核心设计目标是轻量级、易扩展、开发迅速、开箱即用,已有多家公司线上产品使用。该项目于2015年11月发布第一个版本1.0,目前最新版本2.xxx
官网: https://www.xuxueli.com/xxl-job/
GitHub地址: https://github.com/xuxueli/xxl-job/
Gitee地址: https://gitee.com/xuxueli0323/xxl-job/

3. xxl-job框架

xxl-job是一个分布式任务调度平台. 学习博客: https://www.cnblogs.com/newAndHui/p/13862285.html

3.1 架构图

xxl-job架构图
xxl-job架构图
任务触发架构图
任务调度—任务触发—执行器
在这里插入图片描述

3.2 xxl-job安装使用说明

3.2.1 社区支持及源码下载

许雪里社区 https://www.xuxueli.com/page/projects.html
xxl-job的GitHub地址 https://github.com/xuxueli/xxl-job/

3.2.2 初始化admin数据库

初始化任务调用中心admin数据库, sql脚本在源码目录xxl-job/doc/db/下, tables_xxl_job.sql文件.
在这里插入图片描述

3.2.3 修改admin配置文件

主要修改db配置;
在这里插入图片描述
端口 , 上下文名称也可以自己修改.
在这里插入图片描述

3.2.4 访问admin首页

启动admin应用, 访问admin首页http://127.0.0.1:8081/xxl-job-admin,查看执行器列表、任务列表.
用户名 admin
密码 123456

admin首页

3.2.5 新增执行器

执行器管理页面新增执行器.
在这里插入图片描述
新增执行器成功:
在这里插入图片描述

3.2.6 新增任务

任务管理页面新增任务.
在这里插入图片描述
新增任务成功:

在这里插入图片描述

3.2.7 执行器依赖核心包

xxl-job-executor-sample执行器应用中,添加依赖,配置属性,编写任务等;
在这里插入图片描述
1)添加maven依赖

<dependency>
    <groupId>com.xuxueli</groupId>
    <artifactId>xxl-job-core</artifactId>
    <version>${project.parent.version}</version>
</dependency>

2)配置文件信息
db相关配置信息,日志目录配置可以自定义等;

###任务调度中心地址
#### xxl-job admin address list, such as "http://address" or "http://address01,http://address02"
xxl.job.admin.addresses=http://127.0.0.1:8081/xxl-job-admin

#### xxl-job, access token
xxl.job.accessToken=

#### xxl-job executor appname
xxl.job.executor.appname=xxl-job-executor-sample
#### xxl-job executor registry-address: default use address to registry , otherwise use ip:port if address is null
xxl.job.executor.address=
#### xxl-job executor server-info
xxl.job.executor.ip=
xxl.job.executor.port=9998
#### xxl-job executor log-path
xxl.job.executor.logpath=/data/applogs/xxl-job/jobhandler
#### xxl-job executor log-retention-days
xxl.job.executor.logretentiondays=30

3)配置类XxlJobConfig
实例化执行器 XxlJobSpringExecutor

package com.xxl.job.executor.core.config;

import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * xxl-job config
 *
 * @author xuxueli 2017-04-28
 */
@Configuration
public class XxlJobConfig {
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${xxl.job.accessToken}")
    private String accessToken;

    @Value("${xxl.job.executor.appname}")
    private String appname;

    @Value("${xxl.job.executor.address}")
    private String address;

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Value("${xxl.job.executor.port}")
    private int port;

    @Value("${xxl.job.executor.logpath}")
    private String logPath;

    @Value("${xxl.job.executor.logretentiondays}")
    private int logRetentionDays;


    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
        logger.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppname(appname);
        xxlJobSpringExecutor.setAddress(address);
        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
        xxlJobSpringExecutor.setLogPath(logPath);
        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);

        return xxlJobSpringExecutor;
    }

    /**
     * 针对多网卡、容器内部署等情况,可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP;
     *
     *      1、引入依赖:
     *          <dependency>
     *             <groupId>org.springframework.cloud</groupId>
     *             <artifactId>spring-cloud-commons</artifactId>
     *             <version>${version}</version>
     *         </dependency>
     *
     *      2、配置文件,或者容器启动变量
     *          spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'
     *
     *      3、获取IP
     *          String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
     */
}

4)创建handler接口
老版本是创建handler类, 继承了IJobHandler类,重写了execute方法,并被调度执行execute方法。

public class GlueJobHandler extends IJobHandler {
    private long glueUpdatetime;
    private IJobHandler jobHandler;
    public GlueJobHandler(IJobHandler jobHandler, long glueUpdatetime) {
        this.jobHandler = jobHandler;
        this.glueUpdatetime = glueUpdatetime;
    }
    public long getGlueUpdatetime() {
        return glueUpdatetime;
    }

    @Override
    public void execute() throws Exception {
        XxlJobHelper.log("----------- glue.version:"+ glueUpdatetime +" -----------");
        jobHandler.execute();
    }
}

新版本不再是继承IJobHandler,而是编写一个类用注解@Component实例化到容器中,在类中的方法上添加注解@XxlJob("jobName")注解标注这个方法就是要调度执行的任务。

/**
 * XxlJob开发示例(Bean模式)
 *
 * 开发步骤:
 *      1、任务开发:在Spring Bean实例中,开发Job方法;
 *      2、注解配置:为Job方法添加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法")",注解value值对应的是调度中心新建任务的JobHandler属性的值。
 *      3、执行日志:需要通过 "XxlJobHelper.log" 打印执行日志;
 *      4、任务结果:默认任务结果为 "成功" 状态,不需要主动设置;如有诉求,比如设置任务结果为失败,可以通过 "XxlJobHelper.handleFail/handleSuccess" 自主设置任务结果;
 *
 * @author xuxueli 2019-12-11 21:52:51
 */
@Component
public class SampleXxlJob {
    private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);
    /**
     * 1、简单任务示例(Bean模式)
     */
    @XxlJob("demoJobHandler")
    public void demoJobHandler() throws Exception {
        XxlJobHelper.log("XXL-JOB, Hello World.");
        System.out.println("执行了 demoJobHandler --");

        for (int i = 0; i < 5; i++) {
            XxlJobHelper.log("beat at:" + i);
            TimeUnit.SECONDS.sleep(2);
        }
        // default success
    }
}    

4. xxl-job集群部署

4.1 xxl-job-admin集群部署

首先集群部署几台 xxl-job-admin,8080 ,8081端口两台调度中心;
操作:xxl-job-admin.jar 包准备
启动两台任务调度中心:
java -jar xxl-job-admin.jar –server.port=8080
java -jar xxl-job-admin.jar –server.port=8081
访问任务调度平台:
http://localhost:8080/xxl-job-admin/
在这里插入图片描述
http://localhost:8081/xxl-job-admin/
在这里插入图片描述

4.2 xxl-job-executor-example集群部署

4.2.1 集群部署配置

如果xxl-job-admin任务调度中心是集群部署,xxl-job-executor-sample执行器应该通过负载均衡地址进行注册,可以将xxl.job.admin.addresses配置成负载均衡地址或域名;然后分别启动多台执行器应用进行注册。注册之前需要将负载均衡的配置准备好,下面介绍:
执行器注册地址的配置,执行器注册地址使用 域名->nginx代理(负载均衡)->任务调度平台

### 单机
# xxl.job.admin.addresses=http://127.0.0.1:8081/xxl-job-admin
### 集群部署,通过负载均衡的域名地址去注册执行器  域名在hosts文件中配置
xxl.job.admin.addresses=http://www.jobs.com/xxl-job-admin

windows系统hosts文件配置域名,并刷新dns配置;当访问www.jobs.com时就相当于访问的是127.0.0.1的主机地址。当然我们也可以不用域名解析,直接配置主机地址。hosts文件位置 C:\Windows\System32\drivers\etc
在这里插入图片描述
刷新dns配置: ipconfig /flushdns
在这里插入图片描述
Nginx负载均衡拦截配置,当执行器应用向xxl-job-admin注册时,进行负载均衡转发。
在这里插入图片描述
Nginx负载均衡配置,随机转发到两台任务调度中心:admin-8080, admin-8088
如果访问 http://localhost:80/xxl-job-admin 会随机转发到下面地址:
http://localhost:8080/xxl-job-admin
http://localhost:8081/xxl-job-admin
注意,上面的80是nginx监听的端口。
在这里插入图片描述
修改完nginx.conf配置文件后,检查配置文件是否配置正确:

nginx -t -c /softwareInstalled/nginx-1.20.1/conf/nginx.conf

如果nginx已经启动,重新加载配置,并优雅的重启进程

nginx -s reload

如果nginx没启动,那么直接启动Nginx

start nginx

4.2.2 注册执行器

根据上图原理,注册两台注册器到任务调度中心
第一台执行器修改配置并启动:

server.port=8082
xxl.job.admin.addresses=http://www.jobs.com/xxl-job-admin
xxl.job.executor.port=9998

第二台执行器修改配置并启动

server.port=8088
xxl.job.admin.addresses=http://www.jobs.com/xxl-job-admin
xxl.job.executor.port=9999

查看任务调度中心的“执行器管理”, 两台执行器已经注册成功。
http://localhost:8080/xxl-job-admin/jobgroup
在这里插入图片描述
http://localhost:8081/xxl-job-admin/jobgroup
在这里插入图片描述
从执行器注册流程图可以分析出,8082,8088执行器注册到8080和8081调度中心;任务调度中心触发任务时通过9998、9999端口调度执行器去执行任务。(8082、8088端口是执行器对外的web端口)
在这里插入图片描述

4.2.3 任务调度测试

手动执行一次任务,查看执行日志。根据权重转发到了注册到9998的执行器上。
在这里插入图片描述
将注册到9998的执行器停掉。再次手动触发一次任务,注册到9999的执行器依然可以成功执行任务。
在这里插入图片描述
从下面任务调度图可以分析出,8080,8081调度中心是和执行器的9998,9999端口通信的。路由策略也可以自己调整(随机、轮询、指定执行器服务等)。
在这里插入图片描述

5. xxl-job设计思想

  • 将调度行为抽象形成“调度中心”公共平台,而平台自身并不承担业务逻辑,“调度中心”负责发起调度请求。
  • 将任务抽象成分散的JobHandler,交由“执行器”统一管理,“执行器”负责接收调度请求并执行对应的JobHandler中业务逻辑。因此,“调度”和“任务”两部分可以相互解耦,提高系统整体稳定性和扩展性;
  • 调度模块(调度中心):负责管理调度信息,按照调度配置发出调度请求,自身不承担业务代码。调度系统与任务解耦,提高了系统可用性和稳定性,同时调度系统性能不再受限于任务模块;
  • 支持可视化、简单且动态的管理调度信息,包括任务新建、更新、删除、GLUE开发和任务报警等,所有上述操作都会实时生效,同时支持监控调度结果以及执行日志,支持执行器Failover。
  • 执行模块(执行器):负责接收调度请求并执行任务逻辑,任务模块专注于任务的执行等操作,开发和维护更简单和高效;接收“调度中心”的执行请求、终止请求和日志请求等。

6. xxl-job-admin核心代码

6.1 admin初始化工作

1)JobRegistryHelper#getInstance.start()保证任务执行的时候拿到的执行器列表都是运行的,启动一个守护线程。

每30s查询数据库中自动注册的执行器;
查询90s未再次自动注册的执行器;
delete 删除90s未再次注册的执行器register表;
upate 更新group表的addressList;

2)JobRegistryHelper#getInstance.start()启动一个线程,扫描失败日志判断是否需要重试,需要重试则执行触发器;

3)initRpcProvider(); 初始化rpc服务,AdminBiz(任务执行结果回调、注册器执行器、移除执行器)

4)JobScheduleHelper#getInstance.start();

scheduleThread 构建5s内要执行的任务;
ringThread 执行5s内任务的触发器;

6.2 触发器的工作

1)失败次数超过10次的任务使用慢线程池执行;
2)查询要执行任务的详细信息;
3)根据路由策略从执行器地址列表中得到通知的地址;

7. 面试题

你们的项目是分布式的吗?项目定时任务这一块是怎么做的?
分布式任务调度如果让你来实现,有什么思路,怎么实现?
有没有了解过一些开源的方案来解决这一块的问题?

B站视频推荐:
https://www.bilibili.com/video/BV1w541187Wa


文章作者: 王子
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 王子 !
评论
  目录