在第一篇job 的类设计结构中，已经说过job最终执行会在quartz中执行LiteJob该作业，LiteJob中怎样去保证作业的执行的？
 再看一下LiteJob的类图：

LiteJob.png

分析下来，job的执行过程是这张图的样子，比较大：

未命名文件 (3).png

public final class LiteJob implements Job {
    @Setter
    private ElasticJob elasticJob;
    @Setter
    private JobFacade jobFacade;
    
    @Override
    public void execute(final JobExecutionContext context) throws JobExecutionException {
        JobExecutorFactory.getJobExecutor(elasticJob, jobFacade).execute();
    }
}

//接上代码获取执行器
 public static AbstractElasticJobExecutor getJobExecutor(final ElasticJob elasticJob, final JobFacade jobFacade) {
        if (null == elasticJob) {
            return new ScriptJobExecutor(jobFacade);
        }
        if (elasticJob instanceof SimpleJob) {
            return new SimpleJobExecutor((SimpleJob) elasticJob, jobFacade);
        }
        if (elasticJob instanceof DataflowJob) {
            return new DataflowJobExecutor((DataflowJob) elasticJob, jobFacade);
        }
        throw new JobConfigurationException("Cannot support job type '%s'", elasticJob.getClass().getCanonicalName());
    }

 在执行过程中，首先会根据elasticJob的类型(也就是我们在使用elasticJob的过程中，配置的类型)去找到相应的执行器，（ScriptJobExecutor，DataflowJobExecutor，DataflowJobExecutor均实现AbstractElasticJobExecutor接口）。

//AbstractElasticJobExecutor.java 构造方法
 protected AbstractElasticJobExecutor(final JobFacade jobFacade) {
        this.jobFacade = jobFacade;
        jobRootConfig = jobFacade.loadJobRootConfiguration(true);
        jobName = jobRootConfig.getTypeConfig().getCoreConfig().getJobName();
        executorService = ExecutorServiceHandlerRegistry.getExecutorServiceHandler(jobName, (ExecutorServiceHandler) getHandler(JobProperties.JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER));
        jobExceptionHandler = (JobExceptionHandler) getHandler(JobProperties.JobPropertiesEnum.JOB_EXCEPTION_HANDLER);
        itemErrorMessages = new ConcurrentHashMap<>(jobRootConfig.getTypeConfig().getCoreConfig().getShardingTotalCount(), 1);
    }

 从执行器的抽象父类构造方法看，首先会去通过jobFacade然后用configService获取获取job的配置，然后获取一个执行器服务executorService(没有就创建一个executor-service-handler,不配置走默认配置)，再获取异常处理器jobExceptionHandler(作业配置项executor-service-handler，不配置走默认配置)。

 然后看一下job的执行过程：

public final void execute() {
      try ｛
          //检查环境
          jobFacade.checkJobExecutionEnvironment();
      } catch (final JobExecutionEnvironmentException cause) {
          jobExceptionHandler.handleException(jobName, cause);
      }
      //获取分片上下文
      ShardingContexts shardingContexts = jobFacade.getShardingContexts();
      if (shardingContexts.isAllowSendJobEvent()) {
          jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_STAGING, String.format("Job '%s' execute begin.", jobName));
      }
//是否有运行中的任务
      if (jobFacade.misfireIfRunning(shardingContexts.getShardingItemParameters().keySet())) {
          if (shardingContexts.isAllowSendJobEvent()) {
              jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_FINISHED, String.format(
                      "Previous job '%s' - shardingItems '%s' is still running, misfired job will start after previous job completed.", jobName, 
                      shardingContexts.getShardingItemParameters().keySet()));
          }
          return;
      }
      try {
        //通知作业监听对象，作业要开始执行
          jobFacade.beforeJobExecuted(shardingContexts);
          //CHECKSTYLE:OFF
      } catch (final Throwable cause) {
          //CHECKSTYLE:ON
          jobExceptionHandler.handleException(jobName, cause);
      }
      //执行逻辑
      execute(shardingContexts, JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER);
      while (jobFacade.isExecuteMisfired(shardingContexts.getShardingItemParameters().keySet())) {
          jobFacade.clearMisfire(shardingContexts.getShardingItemParameters().keySet());
          execute(shardingContexts, JobExecutionEvent.ExecutionSource.MISFIRE);
      }
      jobFacade.failoverIfNecessary();
      try {
          //执行结束之后，告诉监听器，作业执行结束
          jobFacade.afterJobExecuted(shardingContexts);
          //CHECKSTYLE:OFF
      } catch (final Throwable cause) {
          //CHECKSTYLE:ON
          jobExceptionHandler.handleException(jobName, cause);
      }
  }

 首先检查环境，jobFacade.checkJobExecutionEnvironment();看一下服务器时间与注册中心的时间误差秒数是否在允许范围，配置项：max-time-diff-seconds,-1为不校验时间误差，默认为-1;然后获取分片参数：

  @Override
    public ShardingContexts getShardingContexts() {
        boolean isFailover = configService.load(true).isFailover();
        if (isFailover) {
            List<Integer> failoverShardingItems = failoverService.getLocalFailoverItems();
            if (!failoverShardingItems.isEmpty()) {
                return executionContextService.getJobShardingContext(failoverShardingItems);
            }
        }
        shardingService.shardingIfNecessary();
        List<Integer> shardingItems = shardingService.getLocalShardingItems();
        if (isFailover) {
            shardingItems.removeAll(failoverService.getLocalTakeOffItems());
        }
        shardingItems.removeAll(executionService.getDisabledItems(shardingItems));
        return executionContextService.getJobShardingContext(shardingItems);
    }

  获取分片上下文，首先判断是否执行failOver（失效转移，配置项failOver，默认配置项为false）若分片失效转移为false，则会取判断是否需要分片，做一系列分片逻辑，这里会去加载配置项job-sharding-strategy-class分片策略类，按照策略类分配分片策略，在这里，会去选举主节点，然后从zk更新看是否有上次任务没有做完的情况，有的话会等到上次作业做完，然后重新分片，创建processing节点，再将禁用的分片项去除掉，如果失效转移，则将失效转移的分片项也去除掉。在这里，会去读取配置配置项sharding-total-count,job-parameter, 组装ShardingContexts。
  jobFacade.beforeJobExecuted(shardingContexts);代码是通知监听的listener，看代码：

  
  @Override
    public void beforeJobExecuted(final ShardingContexts shardingContexts) {
        for (ElasticJobListener each : elasticJobListeners) {
            each.beforeJobExecuted(shardingContexts);
        }
    }

  execute(shardingContexts,JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER);这个方法里，根据分片项判断是否有分片，没有分片项，结束掉调度的执行，如果需要向上抛出事件的，抛出已完成事件，结束任务。有分片任务的，去注册作业启动信息，开始执行作业，执行结束之后，将注册信息改为结束状态（改掉JobRegistry的状态和zk的记录）。

private void execute(final ShardingContexts shardingContexts, final JobExecutionEvent.ExecutionSource executionSource) {
        if (shardingContexts.getShardingItemParameters().isEmpty()) {
            if (shardingContexts.isAllowSendJobEvent()) {
                jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_FINISHED, String.format("Sharding item for job '%s' is empty.", jobName));
            }
            return;
        }
        jobFacade.registerJobBegin(shardingContexts);
        String taskId = shardingContexts.getTaskId();
        if (shardingContexts.isAllowSendJobEvent()) {
            jobFacade.postJobStatusTraceEvent(taskId, State.TASK_RUNNING, "");
        }
        try {
            process(shardingContexts, executionSource);
        } finally {
            // TODO 考虑增加作业失败的状态，并且考虑如何处理作业失败的整体回路
          //注册作业的完成
            jobFacade.registerJobCompleted(shardingContexts);
            if (itemErrorMessages.isEmpty()) {
                if (shardingContexts.isAllowSendJobEvent()) {
                    jobFacade.postJobStatusTraceEvent(taskId, State.TASK_FINISHED, "");
                }
            } else {
                //是否发送jobEvent
                if (shardingContexts.isAllowSendJobEvent()) {
                    jobFacade.postJobStatusTraceEvent(taskId, State.TASK_ERROR, itemErrorMessages.toString());
                }
            }
        }
    }

  在registerJobBegin注册作业启动信息的时候，首先改了JobRegistry的作业运行状态，JobRegistry该单例对象维护了所有job的相关信息。其次，如果监控任务执行状态，则创建作业的临时节点。

  /**
     * 注册作业启动信息.
     * 
     * @param shardingContexts 分片上下文
     */
    public void registerJobBegin(final ShardingContexts shardingContexts) {
        JobRegistry.getInstance().setJobRunning(jobName, true);
        if (!configService.load(true).isMonitorExecution()) {
            return;
        }
        for (int each : shardingContexts.getShardingItemParameters().keySet()) {
            jobNodeStorage.fillEphemeralJobNode(ShardingNode.getRunningNode(each), "");
        }
    }

 而在作业的执行过程中，如果作业只有一个分片，则直接去处理作业的请求，如果多于一个，则使用计数器，等所有分片项处理完成再去统一返回，而不是各自分片完成自己的分片任务就返回。

private void process(final ShardingContexts shardingContexts, final JobExecutionEvent.ExecutionSource executionSource) {
        Collection<Integer> items = shardingContexts.getShardingItemParameters().keySet();
        if (1 == items.size()) {
            int item = shardingContexts.getShardingItemParameters().keySet().iterator().next();
            JobExecutionEvent jobExecutionEvent =  new JobExecutionEvent(shardingContexts.getTaskId(), jobName, executionSource, item);
            process(shardingContexts, item, jobExecutionEvent);
            return;
        }
        final CountDownLatch latch = new CountDownLatch(items.size());
        for (final int each : items) {
            final JobExecutionEvent jobExecutionEvent = new JobExecutionEvent(shardingContexts.getTaskId(), jobName, executionSource, each);
            if (executorService.isShutdown()) {
                return;
            }
            executorService.submit(new Runnable() {
                
                @Override
                public void run() {
                    try {
                        process(shardingContexts, each, jobExecutionEvent);
                    } finally {
                        latch.countDown();
                    }
                }
            });
        }
        try {
            latch.await();
        } catch (final InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }

作业请求的处理，会去调用AbstractElasticJobExecutor的process方法，在这个方法里，会直接调用三种基本类型的job的execute方法，也就是我们定义job bean的方法，具体看下面代码：

private void process(final ShardingContexts shardingContexts, final int item, final JobExecutionEvent startEvent) {
        if (shardingContexts.isAllowSendJobEvent()) {
            jobFacade.postJobExecutionEvent(startEvent);
        }
        log.trace("Job '{}' executing, item is: '{}'.", jobName, item);
        JobExecutionEvent completeEvent;
        try {
            //在这里会直接调用三种基本任务的execute方法，
            //该process方法执行的是 AbstractElasticJobExecutor
            //的process抽象方法，具体的实现类可看下面代码 
            process(new ShardingContext(shardingContexts, item));
            completeEvent = startEvent.executionSuccess();
            log.trace("Job '{}' executed, item is: '{}'.", jobName, item);
            if (shardingContexts.isAllowSendJobEvent()) {
                jobFacade.postJobExecutionEvent(completeEvent);
            }
            // CHECKSTYLE:OFF
        } catch (final Throwable cause) {
            // CHECKSTYLE:ON
            completeEvent = startEvent.executionFailure(cause);
            jobFacade.postJobExecutionEvent(completeEvent);
            itemErrorMessages.put(item, ExceptionUtil.transform(cause));
            jobExceptionHandler.handleException(jobName, cause);
        }
    }

//AbstractElasticJobExecutor的实现类
  public final class SimpleJobExecutor extends AbstractElasticJobExecutor {
    
    private final SimpleJob simpleJob;
    
    public SimpleJobExecutor(final SimpleJob simpleJob, final JobFacade jobFacade) {
        super(jobFacade);
        this.simpleJob = simpleJob;
    }
    //process方法实质会调用三种基本任务的execute方法，就是我们配置的作业的执行方法。
    @Override
    protected void process(final ShardingContext shardingContext) {
        simpleJob.execute(shardingContext);
    }
}

jobFacade.failoverIfNecessary();作业执行完成之后，判断是否需要失效转移，再然后 jobFacade.afterJobExecuted(shardingContexts);通知监听的Listenter改作业执行完成。

 @Override
    public void afterJobExecuted(final ShardingContexts shardingContexts) {
        for (ElasticJobListener each : elasticJobListeners) {
            each.afterJobExecuted(shardingContexts);
        }
    }