Elastic-job 分布式任务调度原理「建议收藏」

欢迎大家来到IT世界,在知识的湖畔探索吧!

什么是任务调度？

任务调度是指系统为了自动完成特定任务，在约定的特定的时刻去执行任务的过程。有了任务调度即可解放更多的人力系统自动去执行任务。

任务调度如何实现？

1. 多线程方式实现

实现一个按一定间隔的时间执行任务调度的功能。

简单代码实现：

public static void main(String[] args) {
    // 任务执行的时间间隔
    final long timeInterval = 1000;
    Runable runable = new Runable(){
          public void run() {
               while(true) {
                   try {
                     Thread.sleep(timeInterval);
                   } catch (Exception e){
                      e.printStackTrace();
                   }
               }
          } 
    };
}

欢迎大家来到IT世界,在知识的湖畔探索吧!

JDK也为我们提供了相关的支持，如Timer、ScheduledExecutor

2. Timer方式实现(定时器)

欢迎大家来到IT世界,在知识的湖畔探索吧!public static void main (String[] args) {
  Timer timer = new Timer();
  timer.schedule(new TimerTask(){
        @Override
        public void run(){
           //TODO：业务代码
        }
  }，1000,2000); //1秒后开始调度,每2秒执行一次
}

Timer的优点在于简单易用，每个Timer对应一个线程，因此可以同时启动多个Timer并行执行多个任务，同是一个Timer中的任务是串行执行。

3. ScheduleExecutor方式实现：

public static void main(String[] args) {
  ScheduleExecutorService service = Executors.newScheduledThreadPool(10);
  service.scheduleAtFixedRate(
      new Runable() {
         @Override
         public void run() {
            //TODO:something
            System.out.println("");
         }
      },1,2,TimeUnit.SECONDS);
  )
}

Java5推出了基于线程池设计的ScheduleExecutor，其设计思想是，每一个被调度的任务都会由线程池中一个线程去执行，因此任务是并发执行，相互之间是互不干涉。

4. 第三方的Quartz实现

欢迎大家来到IT世界,在知识的湖畔探索吧!public static void main() {
  // 创建一个Scheduler
  SchedulerFactory schedulerFactory = new StdSchedulerFactory();
  Scheduler scheduler = schedulerFactory.getScheduler();
  // 创建jobDetail
  JobBuilder jobDetailBuilder = JobBuilder,newJob(MyJob.class);
  jobDetailBuilder.withIdentity("jobName","jobGroupName");
  JobDetail jobDetail = jobDetailBuilder.build();
  // 创建触发的CronTrigger 支持按日历调度
  CronTrigger trigger = TriggerBuilder.newTrigger();
        .withIdentity("triggerName","triggerGroupName")
        .startNow()
        .withSchedule(CronScheduleBuilder.cronSchedule("0/2 * * * * ?"))
        .build();
  scheduler.scheduleJob(jobDetail,trigger);
  scheduler.start();
}

什么是分布式？

分布式架构就是将单体的系统架构分为若干个服务，通过网络交互来相互通信。

例如：电商系统为分布式架构，由订单服务，商品服务，用户服务等组成。

分布式的特点:

1. 分布式

2. 伸缩性

3. 高可用

什么是分布式任务调度？

指的是在分布式系统的环境下运行任务调度，我们称之为分布式任务调度

由于采用了分布式架构，一个服务往往会部署多个冗余实例来运行我们的业务。

分布式调度实现的目标：

1. 并行任务调度

并行任务实现靠多线程，如果有大量任务需要调度，此时光靠多线程就会有瓶颈，因为一台计算机CPU的处理能力是有限的。如果将任务调度程序分布式部署，每一个节点还可以部署为集群，这个就可以让多台计算机共同去完成任务调度，我们可以将任务分割为若干个分片，由不同的实例并行执行，来提高任务调度的处理效率。

2. 高可用

若某一个实例宕机，不影响其他的实例来执行任务。

3. 弹性扩容

当集群中增加实例就可以提高并执行任务的处理效率。

4. 任务管理和监测

对系统中存在的所有定时任务进行统一的管理及监测，让开发人员及运维人员能够时刻了解任务执行情况，从而做出快速的应急处理响应。

5. 避免任务重复执行

当任务调度以集群的方式部署，同一个任务可能会执行多次，比如电商系统中到店发优惠券的例子，就会发放多次优惠券，多公司造成很多损失，所以我们需要控制相同的任务在多个运行实例上执行一次，考虑采用下边的方法：

* 分布式锁，多个实例子在任务执行前首先需要获取锁，如果获取失败那么久证明已经有其他的服务已经在运行了，如果获取成功没有服务在运行定时任务，那么就可以执行。

* Zookeeper选举，利用Zookeeper对leader实例执行定时任务，有其他业务已经使用了,那么执行定时任务的时候判断自己是否是leader,如果不是则不执行，如果是则执行业务逻辑，这样也能达到我们的目的。

Elastic-job介绍

Elastic-job是分布式调度的解决方案，由当当网开源，它由两个相互独立的子项目Elastic-job-Lite和Elastic-job-Cloud组成，使用Elastic-job可以快速实现分布式任务调度。

功能列表：

* 分布式调度协调

在分布式环境中，任务能够按指定的调度策略执行，并且能够避免同一个任务多实例重复执行。

* 丰富的调度策略

基于成熟的定时任务作业框架Quartz cron表达式执行定时任务。

* 弹性扩容和缩容

当集群中增加一个实例，它应当也能够被选举并执行任务，当集群减少一个实例时，它所执行的任务能被转移到别的实例来执行。

* 失效转移

某实例在任务执行失败后，能够转移到其他的实例执行。

* 支持并行调度

支持任务分片，任务分片指的是将一个任务分我多个小任务项在多个实例同时执行。

* 作业分片一致性

当任务被分片后，保证同一个分片在分布式环境中仅有一个实例执行。

* 支持作业生命周期操作

可以动态对任务进行开启以及停止操作。

Elastic-job整体的架构？

什么是作业分片？

分片的概念：

作业分片是指任务的分布式执行，需要将一个任务拆分为多个独立的任务项，然后由分布式的应用实例分别执行一个或者几个分片项。

例如：Elastic-job快速入门中文件备份的例子，现有2台服务器，每台服务器分别跑一个应用实例。为了快速的执行作业，那么可以将作业分成4片每个应用实例执行2片，作业遍历数据的逻辑应为：实例1查询text和iamge类型文件执行备份，实例2查找radio和video类型文件备份。如果由于服务器扩容应用实例数量增加为4，则作业遍历数据的逻辑应为：4个实例分别处理text,image,radio,video类型的文件通过对任务的合理分片化，从而达到任务并行处理的效果，最大限度的提高执行作业的吞吐量。

* 分片项与业务处理解耦

Elastic-job并不是直接提供数据处理的功能，框架只会将分片项分配到各个运行中的作业服务器，开发者需要自行处理分片项与真实数据对应的关系。

* 最大限度利用资源

将分片项设置为大于服务器的数量，最好是大于服务器倍数的数量，作业将会合理地利用分布式资源，动态的分配分片项

面试题：Elastic-job分片的原理？

elastic-job的分片是通过zookeeper来实现的，而分片是由主节点分配，如下三种情况都会触发主节点的分片算法执行：

1. 新的job实例加入集群

2. 现有的job实例下线(如果下线的是leader节点，那么先选举然后触发分片算法的执行)

3. 主节点选举

上述三种情况，会让zookeeper上leader节点的sharding节点上多出来一个necessary的临时节点，主节点每次执行job前，都会去看一下这个节点，如果有则执行分片算法。

分片执行结果会存储在zookeeper上，如下图，5个分片，每个分片应该由哪个job实例来运行都已经分配好。分配过程就是上面触发分片算法之后的操作，分配完成之后，各个job实例就会在下次执行的时候使用上这个分配的结果。

每个job实例任务触发前都会获取本任务在本实例上的分片情况，然后封装成shardingContext,传递给调用任务的实际执行方法：

void execute(ShardingContext shardingContext); 

分片算法？

所有的分片策略都继承JobShardingStrategy接口。根据当前注册到ZK的实例列表和在客户端配置的分片数量来进行数据分片。最终将每个Job实例应该获得的分片数字返回出去。 方法签名如下：

/**
     * 作业分片.
     * 
     * @param jobInstances 所有参与分片的单元列表
     * @param jobName 作业名称
     * @param shardingTotalCount 分片总数
     * @return 分片结果
     */
    Map<JobInstance, List<Integer>> sharding(List<JobInstance> jobInstances, String jobName, int shardingTotalCount);

分片函数的触发，只会在leader选举的时候触发，也就是说只会在刚启动和leader节点离开的时候触发，并且是在leader节点上触发，而其他节点不会触发。

1. 基于平均分配算法的分片策略，也就是默认的分片策略

如果分片不能整除，则不能整除的多余分片将依次追加到序号小的服务器。

缺点是：一旦分片数小于作业服务器数，作业将永远分配到ip地址靠前的服务器，导致ip地址靠后的服务器空闲。

2. 根据作业名的哈希值奇偶数决定ip升降序算法的分片策略

作业名的哈希值为奇数则ip升序，作业名的哈希值为偶数则ip降序。

3. 根据作业名的哈希值对服务器列表进行轮转的分片策略

4. 自定义分片策略的原理

主要实现JobShardingStrategy接口，并且把该类配置到分片方法上去。