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

简介: 分布式系列教程(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;
    }
}

总结

相关实践学习
通过日志服务实现云资源OSS的安全审计
本实验介绍如何通过日志服务实现云资源OSS的安全审计。
目录
相关文章
|
10月前
|
消息中间件 运维 Kafka
直播预告|Kafka+Flink双引擎实战:手把手带你搭建分布式实时分析平台!
在数字化转型中,企业亟需从海量数据中快速提取价值并转化为业务增长动力。5月15日19:00-21:00,阿里云三位技术专家将讲解Kafka与Flink的强强联合方案,帮助企业零门槛构建分布式实时分析平台。此组合广泛应用于实时风控、用户行为追踪等场景,具备高吞吐、弹性扩缩容及亚秒级响应优势。直播适合初学者、开发者和数据工程师,参与还有机会领取定制好礼!扫描海报二维码或点击链接预约直播:[https://developer.aliyun.com/live/255088](https://developer.aliyun.com/live/255088)
662 35
直播预告|Kafka+Flink双引擎实战:手把手带你搭建分布式实时分析平台!
|
10月前
|
消息中间件 运维 Kafka
直播预告|Kafka+Flink 双引擎实战:手把手带你搭建分布式实时分析平台!
直播预告|Kafka+Flink 双引擎实战:手把手带你搭建分布式实时分析平台!
311 11
|
6月前
|
负载均衡 算法 调度
基于遗传算法的新的异构分布式系统任务调度算法研究(Matlab代码实现)
基于遗传算法的新的异构分布式系统任务调度算法研究(Matlab代码实现)
277 11
|
6月前
|
消息中间件 监控 Java
Apache Kafka 分布式流处理平台技术详解与实践指南
本文档全面介绍 Apache Kafka 分布式流处理平台的核心概念、架构设计和实践应用。作为高吞吐量、低延迟的分布式消息系统,Kafka 已成为现代数据管道和流处理应用的事实标准。本文将深入探讨其生产者-消费者模型、主题分区机制、副本复制、流处理API等核心机制,帮助开发者构建可靠、可扩展的实时数据流处理系统。
629 4
|
11月前
|
SQL 监控 Go
新一代 Cron-Job分布式调度平台,v1.0.8版本发布,支持Go执行器SDK!
现代化的Cron-Job分布式任务调度平台,支持Go语言执行器SDK,多项核心优势优于其他调度平台。
260 8
|
9月前
|
运维 监控 Linux
WGCLOUD运维平台的分布式计划任务功能介绍
WGCLOUD是一款免费开源的运维监控平台,支持主机与服务器性能监控,具备实时告警和自愈功能。本文重点介绍其计划任务功能模块,可统一管理Linux和Windows主机的定时任务。相比手动配置crontab或Windows任务计划,WGCLOUD提供直观界面,通过添加cron表达式、执行指令或脚本并选择主机,即可轻松完成任务设置,大幅提升多主机任务管理效率。
|
数据采集 监控 数据可视化
11.7K Star!这个分布式爬虫管理平台让多语言协作如此简单!
分布式爬虫管理平台Crawlab,支持任何编程语言和框架的爬虫管理,提供可视化界面、任务调度、日志监控等企业级功能,让爬虫开发管理效率提升300%!
541 1
|
测试技术 调度
新一代 Cron-Job分布式调度平台,v1.0.5版本发布!
增加标签路由能力和多项功能优化!其中Tag标签路由的功能,测试环境多迭代场景下,可通过给任务配置Tag标签,实现任务路由到不同的执行器上。
174 0
|
7月前
|
存储 负载均衡 NoSQL
【赵渝强老师】Redis Cluster分布式集群
Redis Cluster是Redis的分布式存储解决方案,通过哈希槽(slot)实现数据分片,支持水平扩展,具备高可用性和负载均衡能力,适用于大规模数据场景。
520 2
|
7月前
|
存储 缓存 NoSQL
【📕分布式锁通关指南 12】源码剖析redisson如何利用Redis数据结构实现Semaphore和CountDownLatch
本文解析 Redisson 如何通过 Redis 实现分布式信号量(RSemaphore)与倒数闩(RCountDownLatch),利用 Lua 脚本与原子操作保障分布式环境下的同步控制,帮助开发者更好地理解其原理与应用。
469 6

热门文章

最新文章