分布式系列教程(17) - 分布式任务调度平台XXL-JOB

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 分布式系列教程(17) - 分布式任务调度平台XXL-JOB

代码已提交至Github,有兴趣的同学可以下载来看看:https://github.com/ylw-github/SpringBoot-XXLJOB-Demo

本文目录结构:

l____1. 实现任务调度的方案

l________1.1 使用Thread

l________1.2 使用TimerTask

l________1.3 使用ScheduledExecutorService

l________1.4 使用Quartz

l____ 2. 定时任务在分布式下的解决方案

l____ 3. XXLJOB

l________3.1 XXLJOB介绍

l________3.2 XXLJOB集成

l____________3.2.1 部署 xxl-job-admin 注册中心

l____________3.2.2 创建执行器项目

l____________3.2.3 在调度中心中配置执行器

l____________3.2.4 测试

l________3.3 XXLJOB集群

l____总结

1. 实现任务调度的方案

任务调度指的是指定时间去执行指定的任务,那么Java有哪几种实现任务调度的方式呢?直接贴上代码来理解:

1.1 使用Thread

public class ThreadDemo {
  static long count = 0;
  public static void main(String[] args) {
    Runnable runnable = new Runnable() {
      @Override
      public void run() {
        while (true) {
          try {
            Thread.sleep(1000);
            count++;
            System.out.println(count);
          } catch (Exception e) {
            // TODO: handle exception
          }
        }
      }
    };
    Thread thread = new Thread(runnable);
    thread.start();
  }
}

1.2 使用TimerTask

/**
 * 使用TimerTask类实现定时任务
*/
public class TimerTaskDemo {
  static long count = 0;
  public static void main(String[] args) {
    TimerTask timerTask = new TimerTask() {
      @Override
      public void run() {
        count++;
        System.out.println(count);
      }
    };
    Timer timer = new Timer();
    // 天数
    long delay = 0;
    // 秒数
    long period = 1000;
    timer.scheduleAtFixedRate(timerTask, delay, period);
  }
}

1.3 使用ScheduledExecutorService

ScheduledExecutorService是从JavaSE5的java.util.concurrent里,做为并发工具类被引进的,这是最理想的定时任务实现方式。

public class ScheduledExecutorServiceDemo {
  public static void main(String[] args) {
    Runnable runnable = new Runnable() {
      public void run() {
        // task to run goes here
        System.out.println("Hello !!");
      }
    };
    ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
    // 第二个参数为首次执行的延时时间,第三个参数为定时执行的间隔时间
    service.scheduleAtFixedRate(runnable, 1, 1, TimeUnit.SECONDS);
  }
}

1.4 使用Quartz

操作步骤:

1.引入maven依赖

<dependencies>
  <!-- quartz -->
  <dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz</artifactId>
    <version>2.2.1</version>
  </dependency>
  <dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz-jobs</artifactId>
    <version>2.2.1</version>
  </dependency>
</dependencies>

2.任务调度类

public class MyJob implements Job {
  public void execute(JobExecutionContext context) throws JobExecutionException {
    System.out.println("quartz MyJob date:" + new Date().getTime());
  }
}

3.启动类

//1.创建Scheduler的工厂
SchedulerFactory sf = new StdSchedulerFactory();
//2.从工厂中获取调度器实例
Scheduler scheduler = sf.getScheduler();
//3.创建JobDetail
JobDetail jb = JobBuilder.newJob(MyJob.class)
        .withDescription("this is a ram job") //job的描述
        .withIdentity("ramJob", "ramGroup") //job 的name和group
        .build();
//任务运行的时间,SimpleSchedle类型触发器有效
long time=  System.currentTimeMillis() + 3*1000L; //3秒后启动任务
Date statTime = new Date(time);
//4.创建Trigger
//使用SimpleScheduleBuilder或者CronScheduleBuilder
Trigger t = TriggerBuilder.newTrigger()
            .withDescription("")
            .withIdentity("ramTrigger", "ramTriggerGroup")
            //.withSchedule(SimpleScheduleBuilder.simpleSchedule())
            .startAt(statTime)  //默认当前时间启动
            .withSchedule(CronScheduleBuilder.cronSchedule("0/2 * * * * ?")) //两秒执行一次
            .build();
//5.注册任务和定时器
scheduler.scheduleJob(jb, t);
//6.启动 调度器
scheduler.start();

4.Quartz表达式可以参考:

http://cron.qqe2.com/

2. 定时任务在分布式下的解决方案

分布式集群的情况下,怎么保证定时任务不被重复执行?

有如下几个方案:

  1. 使用zookeeper实现分布式锁 (缺点:需要创建临时节点、和事件通知不易于扩展)。
  2. 使用配置文件做一个开关 (缺点:发布后,需要重启)。
  3. 数据库唯一约束( 缺点:效率低)。
  4. 推荐:使用分布式任务调度平台XXLJOBElastric-JobTBSchedule

3. XXLJOB

3.1 XXLJOB介绍

  • 简单:支持通过Web页面对任务进行CRUD操作,操作简单,一分钟上手;
  • 动态:支持动态修改任务状态、暂停/恢复任务,以及终止运行中任务,即时生效;
  • 调度中心HA(中心式) :调度采用中心式设计,“调度中心”基于集群Quartz实现,可保证调度中心HA;
  • 执行器HA(分布式):任务分布式执行,任务"执行器"支持集群部署,可保证任务执行HA;
  • 任务Failover:执行器集群部署时,任务路由策略选择"故障转移"情况下调度失败时将会平滑切换执行器进行Failover;
  • 一致性:“调度中心”通过DB锁保证集群分布式调度的一致性, 一次任务调度只会触发一次执行;
  • 自定义任务参数:支持在线配置调度任务入参,即时生效;
  • 调度线程池:调度系统多线程触发调度运行,确保调度精确执行,不被堵塞;
  • 弹性扩容缩容:一旦有新执行器机器上线或者下线,下次调度时将会重新分配任务
  • 邮件报警:任务失败时支持邮件报警,支持配置多邮件地址群发报警邮件;
  • 状态监控:支持实时监控任务进度;
  • Rolling执行日志:支持在线查看调度结果,并且支持以Rolling方式实时查看执行器输出的完整的执行日志;
  • GLUE:提供Web IDE,支持在线开发任务逻辑代码,动态发布,实时编译生效,省略部署上线的过程。支持30个版本的历史版本回溯。
  • 数据加密:调度中心和执行器之间的通讯进行数据加密,提升调度信息安全性;
  • 任务依赖:支持配置子任务依赖,当父任务执行结束且执行成功后将会主动触发一次子任务的执行, 多个子任务用逗号分隔;
  • 推送maven中央仓库: 将会把最新稳定版推送到maven中央仓库, 方便用户接入和使用;
  • 任务注册: 执行器会周期性自动注册任务, 调度中心将会自动发现注册的任务并触发执行。同时,也支持手动录入执行器地址;
  • 路由策略:执行器集群部署时提供丰富的路由策略,包括:第一个、最后一个、轮询、随机、一致性HASH、最不经常使用、最近最久未使用、故障转移、忙碌转移等;
  • 运行报表:支持实时查看运行数据,如任务数量、调度次数、执行器数量等;以及调度报表,如调度日期分布图,调度成功分布图等;
  • 脚本任务:支持以GLUE模式开发和运行脚本任务,包括Shell、Python等类型脚本;
  • 阻塞处理策略:调度过于密集执行器来不及处理时的处理策略,策略包括:单机串行(默认)、丢弃后续调度、覆盖之前调度;
  • 失败处理策略;调度失败时的处理策略,策略包括:失败告警(默认)、失败重试;
  • 分片广播任务:执行器集群部署时,任务路由策略选择"分片广播"情况下,一次任务调度将会广播触发对应集群中所有执行器执行一次任务,同时传递分片参数;可根据分片参数开发分片任务;
  • 动态分片:分片广播任务以执行器为维度进行分片,支持动态扩容执行器集群从而动态增加分片数量,协同进行业务处理;在进行大数据量业务操作时可显著提升任务处理能力和速度。
  • 事件触发:除了"Cron方式"和"任务依赖方式"触发任务执行之外,支持基于事件的触发任务方式。调度中心提供触发任务单次执行的API服务,可根据业务事件灵活触发。

源码Github地址:https://github.com/xuxueli/xxl-job

文档地址https://www.xuxueli.com/xxl-job/#/

3.2 XXLJOB集成

集成步骤:

1. 部署: xxl-job-admin  作为注册中心
2.创建执行器项目(具体调度地址) 可以支持集群
------ 配置文件需要填写xxl-job注册中心地址
------ 每个具体执行job服务器需要创建一个netty连接端口号
------ 需要执行job的任务类,集成IJobHandler抽象类注册到job容器中
------ Execute方法中编写具体job任务
3.  在调度中心中配置执行器
4. 启动任务

参考:《分布式任务调度平台XXL-JOB搭建教程》

3.2.1 部署 xxl-job-admin 注册中心

1.使用Git Clone源码到IDEA:https://github.com/xuxueli/xxl-job.git

2.执行数据库脚本:/xxl-job/doc/db/tables_xxl_job.sql,可以看到产生了8张表。

3.配置调度中心

调度中心就是源码中的 xxl-job-admin 工程,配置文件地址:/xxl-job/xxl-job-admin/src/main/resources/application.properties

### web
server.port=8080
server.context-path=/xxl-job-admin
### actuator
management.context-path=/actuator
management.health.mail.enabled=false
### resources
spring.mvc.static-path-pattern=/static/**
spring.resources.static-locations=classpath:/static/
### freemarker
spring.freemarker.templateLoaderPath=classpath:/templates/
spring.freemarker.suffix=.ftl
spring.freemarker.charset=UTF-8
spring.freemarker.request-context-attribute=request
spring.freemarker.settings.number_format=0.##########
### mybatis
mybatis.mapper-locations=classpath:/mybatis-mapper/*Mapper.xml
### xxl-job, datasource
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl_job?Unicode=true&characterEncoding=UTF-8
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
spring.datasource.type=org.apache.tomcat.jdbc.pool.DataSource
spring.datasource.tomcat.max-wait=10000
spring.datasource.tomcat.max-active=30
spring.datasource.tomcat.test-on-borrow=true
spring.datasource.tomcat.validation-query=SELECT 1
spring.datasource.tomcat.validation-interval=30000
### xxl-job email
spring.mail.host=smtp.qq.com
spring.mail.port=25
spring.mail.username=xxx@qq.com
spring.mail.password=xxx
spring.mail.properties.mail.smtp.auth=true
spring.mail.properties.mail.smtp.starttls.enable=true
spring.mail.properties.mail.smtp.starttls.required=true
spring.mail.properties.mail.smtp.socketFactory.class=javax.net.ssl.SSLSocketFactory
### xxl-job, access token
xxl.job.accessToken=
### xxl-job, i18n (default empty as chinese, "en" as english)
xxl.job.i18n=
## xxl-job, triggerpool max size
xxl.job.triggerpool.fast.max=200
xxl.job.triggerpool.slow.max=100
### xxl-job, log retention days
xxl.job.logretentiondays=30

4.启动调度中心,启动类在:/xxl-job/xxl-job-admin/src/main/java/com/xxl/job/admin/XxlJobAdminApplication.java

提示日志文件找不到错误:

其实是没有文件目录造成的,直接修改logback文件(目录:/xxl-job/xxl-job-admin/src/main/resources/logback.xml),把日志目录改为当前:

<property name="log.path" value="./xxl-job-admin.log"/>

5.进入调度中心http://localhost:8080/xxl-job-admin/toLogin,登录账号admin,密码123456。

3.2.2 创建执行器项目

1.添加Maven依赖(注意版本号,太低的好像有问题):

<!-- xxl-rpc-core -->
<dependency>
    <groupId>com.xuxueli</groupId>
    <artifactId>xxl-job-core</artifactId>
    <version>2.1.1</version>
</dependency>

2.配置执行器(application.yml):

server:
  #项目端口号
  port: 8081
logging:
  #日志文件
  config: classpath:logback.xml
xxl:
  job:
    admin:
      #调度中心部署跟地址:如调度中心集群部署存在多个地址则用逗号分隔。
      #执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"。
      addresses: http://127.0.0.1:8080/xxl-job-admin
    accessToken:
    #分别配置执行器的名称、ip地址、端口号
    #注意:如果配置多个执行器时,防止端口冲突
    executor:
      appname: executorDemo1
      ip: 127.0.0.1
      port: 9999
      #执行器运行日志文件存储的磁盘位置,需要对该路径拥有读写权限
      logpath: /data/applogs/xxl-job/jobhandler
      #执行器Log文件定期清理功能,指定日志保存天数,日志文件过期自动删除。限制至少保持3天,否则功能不生效;
      #-1表示永不删除
      logretentiondays: -1

3.载入配置文件(新增XxlJobConfig文件):

package com.ylw.job.executor.config;
import com.xxl.job.core.executor.XxlJobExecutor;
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.ComponentScan;
import org.springframework.context.annotation.Configuration;
@Configuration
@ComponentScan(basePackages = "com.ylw.job.executor.jobhandler")
public class XxlJobConfig {
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);
    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;
    @Value("${xxl.job.executor.appname}")
    private String appName;
    @Value("${xxl.job.executor.ip}")
    private String ip;
    @Value("${xxl.job.executor.port}")
    private int port;
    @Value("${xxl.job.accessToken}")
    private String accessToken;
    @Value("${xxl.job.executor.logpath}")
    private String logPath;
    @Value("${xxl.job.executor.logretentiondays}")
    private int logRetentionDays;
    @Bean(initMethod = "start", destroyMethod = "destroy")
    public XxlJobExecutor xxlJobExecutor() {
        logger.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobExecutor xxlJobExecutor = new XxlJobExecutor();
        xxlJobExecutor.setAdminAddresses(adminAddresses);
        xxlJobExecutor.setAppName(appName);
        xxlJobExecutor.setIp(ip);
        xxlJobExecutor.setPort(port);
        xxlJobExecutor.setAccessToken(accessToken);
        xxlJobExecutor.setLogPath(logPath);
        xxlJobExecutor.setLogRetentionDays(logRetentionDays);
        return xxlJobExecutor;
    }
}

4.创建任务JobHandler

@JobHandler("everySecondsHandler")
@Component
public class EverySecondsHandler extends IJobHandler {
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);
    @Value("${server.port}")
    private String serverPort;
    @Override
    public ReturnT<String> execute(String param) throws Exception {
        logger.info("######端口号:serverPort" + serverPort + "###定时Job开始执行啦!!!!######");
        return SUCCESS;
    }
}
3.2.3 在调度中心中配置执行器

1.新增执行器:

  • AppName:是每个执行器集群的唯一标示AppName, 执行器会周期性以AppName为对象进行自动注册。可通过该配置自动发现注册成功的执行器, 供任务调度时使用;
  • 名称:执行器的名称, 因为AppName限制字母数字等组成,可读性不强, 名称为了提高执行器的可读性;
  • 排序: 执行器的排序, 系统中需要执行器的地方,如任务新增, 将会按照该排序读取可用的执行器列表;
  • 注册方式:调度中心获取执行器地址的方式,
  • 自动注册:执行器自动进行执行器注册,调度中心通过底层注册表可以动态发现执行器机器地址;
  • 手动录入:人工手动录入执行器的地址信息,多地址逗号分隔,供调度中心使用;
  • 机器地址:"注册方式"为"手动录入"时有效,支持人工维护执行器的地址信息;

2.新增任务

3.2.4 测试

1.启动SpringBoot项目

2.控制台启动任务:

可以看到日志打印:

3.3 XXL-JOB集群

详情可以参考博客:《XXL-Job高可用集群搭建》

  • 调度中心支持集群部署,提升调度系统容灾和可用性。
  • 调度中心集群部署时,几点要求和建议:
  • DB配置保持一致;
  • 登陆账号配置保持一致;
  • 群机器时钟保持一致(单机集群忽视);

建议:推荐通过nginx为调度中心集群做负载均衡,分配域名。调度中心访问、执行器回调配置、调用API服务等操作均通过该域名进行。

Nginx配置信息:

upstream  backServer{
  server 127.0.0.1:8080 weight=1;
  server 127.0.0.1:8081 weight=1;
}
server {
    listen       80;
    server_name  www.xxx.com;
    #charset koi8-r;
    #access_log  logs/host.access.log  main;
    location / {
        proxy_pass   http://backServer;
        index  index.html index.htm;
    }
    error_page   500 502 503 504  /50x.html;
    location = /50x.html {
        root   html;
    }
}

总结

相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
目录
相关文章
|
4月前
|
存储 监控 固态存储
【vSAN分布式存储服务器数据恢复】VMware vSphere vSAN 分布式存储虚拟化平台VMDK文件1KB问题数据恢复案例
在一例vSAN分布式存储故障中,因替换故障闪存盘后磁盘组失效,一台采用RAID0策略且未使用置备的虚拟机VMDK文件受损,仅余1KB大小。经分析发现,该VMDK文件与内部虚拟对象关联失效导致。恢复方案包括定位虚拟对象及组件的具体物理位置,解析分配空间,并手动重组RAID0结构以恢复数据。此案例强调了深入理解vSAN分布式存储机制的重要性,以及定制化数据恢复方案的有效性。
107 5
|
1月前
|
消息中间件 监控 数据可视化
Apache Airflow 开源最顶级的分布式工作流平台
Apache Airflow 是一个用于创作、调度和监控工作流的平台,通过将工作流定义为代码,实现更好的可维护性和协作性。Airflow 使用有向无环图(DAG)定义任务,支持动态生成、扩展和优雅的管道设计。其丰富的命令行工具和用户界面使得任务管理和监控更加便捷。适用于静态和缓慢变化的工作流,常用于数据处理。
Apache Airflow 开源最顶级的分布式工作流平台
|
5月前
|
机器学习/深度学习 人工智能 Shell
人工智能平台PAI操作报错合集之在分布式训练过程中遇到报错,是什么原因
阿里云人工智能平台PAI是一个功能强大、易于使用的AI开发平台,旨在降低AI开发门槛,加速创新,助力企业和开发者高效构建、部署和管理人工智能应用。其中包含了一系列相互协同的产品与服务,共同构成一个完整的人工智能开发与应用生态系统。以下是对PAI产品使用合集的概述,涵盖数据处理、模型开发、训练加速、模型部署及管理等多个环节。
|
3月前
|
存储 NoSQL 调度
|
2月前
|
SQL NoSQL MongoDB
一款基于分布式文件存储的数据库MongoDB的介绍及基本使用教程
一款基于分布式文件存储的数据库MongoDB的介绍及基本使用教程
51 0
|
4月前
|
消息中间件 Java Kafka
"Kafka快速上手:从环境搭建到Java Producer与Consumer实战,轻松掌握分布式流处理平台"
【8月更文挑战第10天】Apache Kafka作为分布式流处理平台的领头羊,凭借其高吞吐量、可扩展性和容错性,在大数据处理、实时日志收集及消息队列领域表现卓越。初学者需掌握Kafka基本概念与操作。Kafka的核心组件包括Producer(生产者)、Broker(服务器)和Consumer(消费者)。Producer发送消息到Topic,Broker负责存储与转发,Consumer则读取这些消息。首先确保已安装Java和Kafka,并启动服务。接着可通过命令行创建Topic,并使用提供的Java API实现Producer发送消息和Consumer读取消息的功能。
87 8
|
5月前
|
负载均衡 监控 搜索推荐
面试题ES问题之Solr和Elasticsearch在分布式管理上如何解决
面试题ES问题之Solr和Elasticsearch在分布式管理上如何解决
43 1
|
5月前
|
存储 缓存 分布式计算
高并发架构设计三大利器:缓存、限流和降级问题之缓存的应对策略问题如何解决
高并发架构设计三大利器:缓存、限流和降级问题之缓存的应对策略问题如何解决
|
2月前
|
NoSQL Java Redis
太惨痛: Redis 分布式锁 5个大坑,又大又深, 如何才能 避开 ?
Redis分布式锁在高并发场景下是重要的技术手段,但其实现过程中常遇到五大深坑:**原子性问题**、**连接耗尽问题**、**锁过期问题**、**锁失效问题**以及**锁分段问题**。这些问题不仅影响系统的稳定性和性能,还可能导致数据不一致。尼恩在实际项目中总结了这些坑,并提供了详细的解决方案,包括使用Lua脚本保证原子性、设置合理的锁过期时间和使用看门狗机制、以及通过锁分段提升性能。这些经验和技巧对面试和实际开发都有很大帮助,值得深入学习和实践。
太惨痛: Redis 分布式锁 5个大坑,又大又深, 如何才能 避开 ?