datax(12):调度源码解读AbstractScheduler「建议收藏」

大家好，又见面了，我是你们的朋友全栈君。

datax的jobContainer最终会通过调度周期性的执行，今天把它看完；

一、基类AbstractScheduler概述

类继承关系

全部方法

二、AbstractScheduler的主要属性和方法

1、主要属性

  /** * 脏数据行数检查器，用于运行中随时检查脏数据是否超过限制（脏数据行数，或脏数据百分比） */
  private ErrorRecordChecker errorLimit;

  /** * 积累容器通讯器，来处理JobContainer、TaskGroupContainer和Task的通讯 */
  private AbstractContainerCommunicator containerCommunicator;

2、主要方法

  /** * 默认调度执行方法 <br> * 1 传入多个调度配置，获取报告时间+休息时间+jobId（赋值给全局jobId），生成错误记录检查类 * 2 给全局jobId赋值，生成错误记录检查类，生成容器通讯类（反馈任务信息） * 3 根据入参计算task的数量，开始所有taskGroup * * @param cfg List<Configuration> */
  public void schedule(List<Configuration> cfg) { 
   
    xxx
}
  /** * 开始所有的taskGroup，只允许本包的类访问 * * @param configurations List<Configuration> */
  protected abstract void startAllTaskGroup(List<Configuration> configurations);
 

三、谁调用AbstractScheduler的schedule

从JobContainer.schedule调用AbstractScheduler.schedule

四、schedule和startAllTaskGroup方法解析

schedule方法主要在AbstractScheduler实现
运行时序图

  /** * 默认调度执行方法 <br> * 1 传入多个调度配置，获取报告时间+休息时间+jobId（赋值给全局jobId），生成错误记录检查类 * 2 给全局jobId赋值，生成错误记录检查类，生成容器通讯类（反馈任务信息） * 3 根据入参计算task的数量，开始所有taskGroup * * @param cfg List<Configuration> */
  public void schedule(List<Configuration> cfg) { 
   
    Validate.notNull(cfg, "scheduler配置不能为空");
    int reportMillSec = cfg.get(0).getInt(DATAX_CORE_CONTAINER_JOB_REPORTINTERVAL, 30000);
    int sleepMillSec = cfg.get(0).getInt(DATAX_CORE_CONTAINER_JOB_SLEEPINTERVAL, 10000);

    this.jobId = cfg.get(0).getLong(CoreConstant.DATAX_CORE_CONTAINER_JOB_ID);
    errorLimit = new ErrorRecordChecker(cfg.get(0));
    //给 taskGroupContainer 的 Communication 注册
    this.containerCommunicator.registerCommunication(cfg);
    int taskCnt = calculateTaskCount(cfg);
    startAllTaskGroup(cfg);
    Communication lastComm = new Communication();
    long lastReportTimeStamp = System.currentTimeMillis();
    try { 
   
      while (true) { 
   
        /** * step 1: collect job stat * step 2: getReport info, then report it * step 3: errorLimit do check * step 4: dealSucceedStat(); * step 5: dealKillingStat(); * step 6: dealFailedStat(); * step 7: refresh last job stat, and then sleep for next while * * above steps, some ones should report info to DS * */
        Communication nowComm = this.containerCommunicator.collect();
        nowComm.setTimestamp(System.currentTimeMillis());
        LOG.debug(nowComm.toString());

        //汇报周期
        long now = System.currentTimeMillis();
        if (now - lastReportTimeStamp > reportMillSec) { 
   
          Communication comm = CommunicationTool.getReportCommunication(nowComm, lastComm, taskCnt);

          this.containerCommunicator.report(comm);
          lastReportTimeStamp = now;
          lastComm = nowComm;
        }

        errorLimit.checkRecordLimit(nowComm);
        if (nowComm.getState() == State.SUCCEEDED) { 
   
          LOG.info("Scheduler accomplished all tasks.");
          break;
        }

        if (isJobKilling(this.getJobId())) { 
   
          dealKillingStat(this.containerCommunicator, taskCnt);
        } else if (nowComm.getState() == State.FAILED) { 
   
          dealFailedStat(this.containerCommunicator, nowComm.getThrowable());
        }
        Thread.sleep(sleepMillSec);
      }
    } catch (InterruptedException e) { 
   
      // 以 failed 状态退出
      LOG.error("捕获到InterruptedException异常!", e);
      throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR, e);
    }
  }

startAllTaskGroup方法在ProcessInnerScheduler实现
运行时序图

  /** * 1、创建线程池 <br/> * 2、变量传入的cfgs，生成tgRunner，然后线程池执行 <br/> * 3、线程池关闭 <br/> * * @param cfgs List<Configuration> */
  @Override
  public void startAllTaskGroup(List<Configuration> cfgs) { 
   
    this.taskGroupContainerExecutorService = new ThreadPoolExecutor(cfgs.size(), cfgs.size(),
        0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
    for (Configuration taskGroupCfg : cfgs) { 
   
      TaskGroupContainerRunner taskGroupContainerRunner = newTaskGroupContainerRunner(taskGroupCfg);
      this.taskGroupContainerExecutorService.execute(taskGroupContainerRunner);
    }
    this.taskGroupContainerExecutorService.shutdown();
  }

注：

1. 对源码进行略微改动，主要修改为 1 阿里代码规约扫描出来的，2 clean code；

2. 所有代码都已经上传到github（master分支和dev），可以免费白嫖