定时任务

定时任务

• 在项目开发中经常需要指定一些任务，比如：在每天的凌晨时分析一天日志。再比如每天凌晨3点进行数据库

  的备份

• 可以使用：

  • timer
  • scheduledThreadPoolExecutor
  • quartz框架
  • spring task框架
  • xxl-job框架

  Timer

  @Test
  public void testTimer(){
  
    Timer timer  = new Timer();
    timer.schedule(new TimerTask() {
        @Override
        public void run() {
            System.out.println("定时任务一");
        }
    },2000,1000);
  
    new Scanner(System.in).next();
  
  }

  Timer缺陷

  • Timer每次都创建了一个新的线程。
  • Timer创建的线程没有处理异常，因此一旦抛出非受检异常，该线程会立即终止。
  • 定时只能粗略以固定值进行，没法实现精确定时任务
  • JDK 8.0以后推荐使用ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor

@Test
public void test(){
    ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(20);
    executor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+":定时任务二");
        }
    },2000,1000, TimeUnit.MILLISECONDS);

    new Scanner(System.in).next();
}

ScheduledThreadPoolExecutor缺陷

• ScheduledThreadPoolExecutor创建的线程没有处理异常，因此一旦抛出非受检异常，该线程会立即终止。
• 定时只能粗略以固定值进行，没法实现精确定时任务

Quartz框架

• 七子表达式=>表达式包含七段： 秒 分 时 日 月 周 年，每段间用空格隔开

• 特殊字符

• 示例

// 每天凌晨零点执行
@Scheduled(cron ="0 0 0 * * * ?")
// 每隔五分钟执行
@Scheduled(cron ="0 */5 * * * * ?")
// 在每天下午2点到下午2:55期间的每5分钟触发
@Scheduled(cron ="0 0/5 14 * * ?")
// 在每天下午2点到2:55期间和下午6点到6:55期间的每5分钟触发
@Scheduled(cron ="0 0/5 14,18 * * ?")
// 在每天下午2点到下午2:05期间的每1分钟触发
@Scheduled(cron ="0 0-5 14 * * ?")
// 每个星期三中午12点
@Scheduled(cron ="0 0 12 ? * WED")
// 每月的第三个星期五上午10:15触发
@Scheduled(cron ="0 15 10 ? * 6#3")
// 每年三月的星期三的下午2:10和2:44触发
@Scheduled(cron ="0 10,44 14 ? 3 WED")
// 每月最后一日的上午10:15触发
@Scheduled(cron ="0 15 10 L * ?")
// 每月的最后一个星期五上午10:15触发
@Scheduled(cron ="0 15 10 ? * 6L")

• 在线生成 : http://cron.qqe2.com/
• 示例

<!-- https://mvnrepository.com/artifact/quartz/quartz -->
<dependency>
    <groupId>quartz</groupId>
    <artifactId>quartz</artifactId>
    <version>1.5.2</version>
</dependency>

 @Test
public void testQuartz(){
    //通过SchedulerFactory获得一个调度器
    SchedulerFactory schedulerfactory = new StdSchedulerFactory();

    try {
        Scheduler scheduler = schedulerfactory.getScheduler();
        // 指明job的名称，所在组的名称，以及绑定job类    创建jobDetail实例，绑定Job实现类
        JobDetail jobDetail = new JobDetail("job1","jobGroup1",MyJob.class);

        CronTrigger cronTrigger = new CronTrigger("cronName","conGroup");
        cronTrigger.setCronExpression("0/3 * * * * ? ");
        scheduler.scheduleJob(jobDetail,cronTrigger);
        scheduler.start();
    } catch (SchedulerException e) {
        e.printStackTrace();
    } catch (ParseException e) {
        e.printStackTrace();
    }

    new Scanner(System.in).next();

}

public class MyJob implements Job {
    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        System.out.println("定时任务3");
    }
}

Quartz的优点

• Quartz创建的线程没有处理异常，因此一旦抛出非受检异常，定时任务不会终止。
• 通过七子表达式实现精确定时任务

Spring Task框架

• 在标签@Sheduled注解的时候需要指定cron表达式
• 表达式包含六段： 秒 分 时 日 月 周几，每段间用空格隔开
• 字段介绍

两步曲

定制任务 @Scheduled

@Service
public class TaskService {
    @Scheduled(cron = "0/4 * * * * ?" )
    public void job1() { 
        System.out.println("任务进行中。。。");
    } 
}

启用定时任务

@EnableScheduling
public class Application{

}

Spring Task 缺点：

• 做集群任务的重复执行问题
• cron表达式定义在代码之中，修改不方便
• 定时任务失败了，无法重试也没有统计
• 如果任务量过大，不能有效的分片执行

分布式任务调度

因为单机任务调度，存在上述的问题，因而需要分布式任务调度，针对分布式任务调度的需求，市场上出现了很多的产品：

• TBSchedule：淘宝推出的一款非常优秀的高性能分布式调度框架，目前被应用于阿里、京东、支付宝、国美等很多互联网企业的流程调度系统中。但是已经多年未更新，文档缺失严重，缺少维护。

• XXL-Job：大众点评的分布式任务调度平台，是一个轻量级分布式任务调度平台, 其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。

• Elastic-job：当当网借鉴TBSchedule并基于quartz 二次开发的弹性分布式任务调度系统，功能丰富强大，采用zookeeper实现分布式协调，具有任务高可用以及分片功能。

• Saturn： 唯品会开源的一个分布式任务调度平台，基于Elastic-job，可以全域统一配置，统一监 控，具有任务高可用以及分片功能

下面的章节内容就以XXL-JOB为例，重点学习下XXL-JOB的如何使用。

XXL-JOB框架

概述

XXL-JOB是一个分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。

源码地址

XXL-JOB是一个分布式定时器任务派遣服务，这个项目主要有以下三部分组成 ：

• xxl-job-admin： 调度中心，主要是提供任务管理平台的页面，需要把该模块单独打包作为一个服务部署，定时器再执行时，是通过该服务去调用我们的业务服务完成任务执行。推荐使用docker部署
• xxl-job-core: 公共依赖模块，在整合业务服务的时候，需要引入该依赖。推荐Maven依赖
• xxl-job-executor-samples：官方提供的demo。可以学习参考

初始化调度数据库

位置：/xxl-job/doc/db/tables_xxl_job.sql 共8张表

• xxl_job_lock：任务调度锁表；
• xxl_job_info：调度扩展信息表： 用于保存XXL-JOB调度任务的扩展信息，如任务分组、任务名、机器地址、执行器、执行入参和报警邮件等等；
• xxl_job_log：调度日志表： 用于保存XXL-JOB任务调度的历史信息，如调度结果、执行结果、调度入参、调度机器和执行器等等；
• xxl_job_logglue：任务GLUE日志：用于保存GLUE更新历史，用于支持GLUE的版本回溯功能；
• xxl_job_registry：执行器注册表，维护在线的执行器和调度中心机器地址信息；
• xxl_job_user：系统用户表；

调度中心支持集群部署，集群情况下各节点务必连接同一个mysql实例;

如果mysql做主从,调度中心集群节点务必强制走主库;

安装

docker-compose.yaml

  xxl-job-admin:
    image: xuxueli/xxl-job-admin:2.4.1
    container_name: xxl-job-admin
    privileged: true
    restart: always
    ports:
      - 8868:8080
    environment:
      TZ: Asia/Shanghai
      PARAMS: '
      --server.servlet.context-path=/xxl-job-admin
      --spring.datasource.url=jdbc:mysql://mysql:3306/xxl_job?Unicode=true&characterEncoding=UTF-8 
      --spring.datasource.username=root 
      --spring.datasource.password=root 
      --xxl.job.accessToken=xxl-job
      --xxl.job.login.username=root 
      --xxl.job.login.password=root'
    volumes:
      - /docker/xxl-job-admin/xxl-job:/xxl-job
      - /docker/xxl-job-admin/applogs:/data/applogs
    depends_on:
      - mysql # 要依赖mysql，前提是mysql已安装

访问：http://192.168.37.252:8868/xxl-job-admin/toLogin

账户/密码: admin/123456

示例

以SpringBoot为例

pom.xml

boot-web
lombok
xxl-job-core

application.yml

spring:
  application:
    name: my
xxl:
  job:
    # 执行器开关
    enabled: true
    # 调度中心TOKEN
    accessToken: xxl-job
    # 调度中心地址：如调度中心集群部署存在多个地址则用逗号分隔。【服务器IP】
    admin-addresses: http://192.168.69.252:8868/xxl-job-admin
    # 调度中心应用名 通过服务名连接调度中心(启用admin-appname会导致admin-addresses不生效)
    # admin-appname: wanlimall-xxl-job-admin
    # 执行器配置
    executor:
      # 执行器AppName
      appname: ${spring.application.name}-executor
      # 执行器IP：默认自动获取IP，有多网卡的机器推荐手动填写 [选填]【项目本地IP】
      ip: 192.168.69.251
      #执行器端口号: 小于等于0则自动获取 默认端口为9999，单机部署多个执行器时，注意要配置不同执行器端口；
      port: 9999
      # 执行器注册: 默认IP:PORT [选填]
      address:
      # 执行器运行日志文件存储磁盘路径 [选填] 需要对该路径拥有读写权限；为空则使用默认路径；
      logpath: D:/xxl-job
      # 执行器日志文件保存天数 [选填] 过期日志自动清理, 限制值大于等于3时生效  否则, 如-1, 关闭自动清理功能
      logretentiondays: 30

以下是执行器的属性说明：

XxlJobProperties.java

@Data
@ConfigurationProperties(prefix = "xxl.job")
public class XxlJobProperties {

    private Boolean enabled;

    private String adminAddresses;

    private String adminAppname;

    private String accessToken;

    private Executor executor;

    @Data
    @NoArgsConstructor
    public static class Executor {

        private String appname;

        private String address;

        private String ip;

        private int port;

        private String logPath;

        private int logRetentionDays;
    }
}

XxlJobConfig.java

单体版

@Configuration
@EnableConfigurationProperties(XxlJobProperties.class)
@ConditionalOnProperty(prefix = "xxl.job", name = "enabled", havingValue = "true")
public class XxlJobConfig {

    @Resource
    private  XxlJobProperties xxlJobProperties;

    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(xxlJobProperties.getAdminAddresses());
        xxlJobSpringExecutor.setAccessToken(xxlJobProperties.getAccessToken());

        XxlJobProperties.Executor executor = xxlJobProperties.getExecutor();
        xxlJobSpringExecutor.setAppname(executor.getAppname());
        xxlJobSpringExecutor.setAddress(executor.getAddress());
        xxlJobSpringExecutor.setIp(executor.getIp());
        xxlJobSpringExecutor.setPort(executor.getPort());
        xxlJobSpringExecutor.setLogPath(executor.getLogPath());
        xxlJobSpringExecutor.setLogRetentionDays(executor.getLogRetentionDays());
        return xxlJobSpringExecutor;
    }
}

微服务版

@Slf4j
@AutoConfiguration
@EnableConfigurationProperties(XxlJobProperties.class)
@AllArgsConstructor
@ConditionalOnProperty(prefix = "xxl.job", name = "enabled", havingValue = "true")
public class XxlJobConfig {

    private final XxlJobProperties xxlJobProperties;

    private final DiscoveryClient discoveryClient;

    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
        log.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        if (StringUtils.isNotBlank(xxlJobProperties.getAdminAppname())) {
            List<ServiceInstance> instances = discoveryClient.getInstances(xxlJobProperties.getAdminAppname());
            if (CollUtil.isEmpty(instances)) {
                throw new RuntimeException("调度中心不存在!");
            }
            String serverList = StreamUtils.join(instances, instance ->
                String.format("http://%s:%s", instance.getHost(), instance.getPort()));
            xxlJobSpringExecutor.setAdminAddresses(serverList);
        } else {
            xxlJobSpringExecutor.setAdminAddresses(xxlJobProperties.getAdminAddresses());
        }
        xxlJobSpringExecutor.setAccessToken(xxlJobProperties.getAccessToken());
        XxlJobProperties.Executor executor = xxlJobProperties.getExecutor();
        xxlJobSpringExecutor.setAppname(executor.getAppname());
        xxlJobSpringExecutor.setAddress(executor.getAddress());
        xxlJobSpringExecutor.setIp(executor.getIp());
        xxlJobSpringExecutor.setPort(executor.getPort());
        xxlJobSpringExecutor.setLogPath(executor.getLogPath());
        xxlJobSpringExecutor.setLogRetentionDays(executor.getLogRetentionDays());
        return xxlJobSpringExecutor;
    }

}

Task

@Component
public class TestTask {

    @XxlJob("hello")
    public void demoJobHandler() throws Exception {
        System.out.println("hello---->xxl-job");
    }
}

调试中心页面

执行器管理

任务管理

基础配置参数

• 执行器：每个任务必须绑定一个执行器, 方便给任务进行分组

• 任务描述：任务的描述信息，便于任务管理；

• 负责人：任务的负责人；

• 报警邮件：任务调度失败时邮件通知的邮箱地址，支持配置多邮箱地址，配置多个邮箱地址时用逗号分隔

• 调度类型：

  • 无：该类型不会主动触发调度；
  • CRON：该类型将会通过CRON，触发任务调度；
  • 固定速度：该类型将会以固定速度，触发任务调度；按照固定的间隔时间，周期性触发；

• 运行模式：

• JobHandler：运行模式为 "BEAN模式" 时生效，对应执行器中新开发的JobHandler类“@JobHandler”注解自定义的value值；

• 执行参数：任务执行所需的参数；

• BEAN模式：任务以JobHandler方式维护在执行器端；需要结合 "JobHandler" 属性匹配执行器中任务；

• 阻塞处理策略

• 阻塞处理策略：调度过于密集执行器来不及处理时的处理策略；

• 单机串行（默认）：调度请求进入单机执行器后，调度请求进入FIFO(First Input First Output)队列并以串行方式运行；

• 丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败；

• 覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务；

• 路由策略

  当执行器集群部署时，提供丰富的路由策略，包括；

• FIRST（第一个）：固定选择第一个机器；

• LAST（最后一个）：固定选择最后一个机器；

• ROUND（轮询）

• RANDOM（随机）：随机选择在线的机器；

• CONSISTENT_HASH（一致性HASH）：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。

• LEAST_FREQUENTLY_USED（最不经常使用）：使用频率最低的机器优先被选举；

• LEAST_RECENTLY_USED（最近最久未使用）：最久未使用的机器优先被选举；

• FAILOVER（故障转移）：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度；

• BUSYOVER（忙碌转移）：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度；

• SHARDING_BROADCAST(分片广播)：广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务；