?在第一篇job 的類設(shè)計(jì)結(jié)構(gòu)中围来，已經(jīng)說(shuō)過(guò)job最終執(zhí)行會(huì)在quartz中執(zhí)行LiteJob該作業(yè)靠柑，LiteJob中怎樣去保證作業(yè)的執(zhí)行的？
?再看一下LiteJob的類圖：

LiteJob.png

分析下來(lái)梯轻，job的執(zhí)行過(guò)程是這張圖的樣子美浦，比較大：

未命名文件 (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();
    }
}

//接上代碼獲取執(zhí)行器
 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());
    }

?在執(zhí)行過(guò)程中，首先會(huì)根據(jù)elasticJob的類型(也就是我們?cè)谑褂胑lasticJob的過(guò)程中戚哎，配置的類型)去找到相應(yīng)的執(zhí)行器裸诽，（ScriptJobExecutor，DataflowJobExecutor型凳，DataflowJobExecutor均實(shí)現(xiàn)AbstractElasticJobExecutor接口）崭捍。

//AbstractElasticJobExecutor.java 構(gòu)造方法
 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);
    }

?從執(zhí)行器的抽象父類構(gòu)造方法看，首先會(huì)去通過(guò)jobFacade然后用configService獲取獲取job的配置啰脚，然后獲取一個(gè)執(zhí)行器服務(wù)executorService(沒(méi)有就創(chuàng)建一個(gè)executor-service-handler,不配置走默認(rèn)配置)殷蛇，再獲取異常處理器jobExceptionHandler(作業(yè)配置項(xiàng)executor-service-handler，不配置走默認(rèn)配置)橄浓。

?然后看一下job的執(zhí)行過(guò)程：

public final void execute() {
      try ｛
          //檢查環(huán)境
          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));
      }
//是否有運(yùn)行中的任務(wù)
      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 {
        //通知作業(yè)監(jiān)聽(tīng)對(duì)象粒梦，作業(yè)要開(kāi)始執(zhí)行
          jobFacade.beforeJobExecuted(shardingContexts);
          //CHECKSTYLE:OFF
      } catch (final Throwable cause) {
          //CHECKSTYLE:ON
          jobExceptionHandler.handleException(jobName, cause);
      }
      //執(zhí)行邏輯
      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 {
          //執(zhí)行結(jié)束之后，告訴監(jiān)聽(tīng)器荸实，作業(yè)執(zhí)行結(jié)束
          jobFacade.afterJobExecuted(shardingContexts);
          //CHECKSTYLE:OFF
      } catch (final Throwable cause) {
          //CHECKSTYLE:ON
          jobExceptionHandler.handleException(jobName, cause);
      }
  }

?首先檢查環(huán)境匀们，jobFacade.checkJobExecutionEnvironment();看一下服務(wù)器時(shí)間與注冊(cè)中心的時(shí)間誤差秒數(shù)是否在允許范圍，配置項(xiàng)：max-time-diff-seconds,-1為不校驗(yàn)時(shí)間誤差准给，默認(rèn)為-1;然后獲取分片參數(shù)：

  @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);
    }

??獲取分片上下文泄朴，首先判斷是否執(zhí)行failOver（失效轉(zhuǎn)移，配置項(xiàng)failOver露氮，默認(rèn)配置項(xiàng)為false）若分片失效轉(zhuǎn)移為false祖灰，則會(huì)取判斷是否需要分片，做一系列分片邏輯畔规，這里會(huì)去加載配置項(xiàng)job-sharding-strategy-class分片策略類局扶，按照策略類分配分片策略，在這里叁扫，會(huì)去選舉主節(jié)點(diǎn)三妈，然后從zk更新看是否有上次任務(wù)沒(méi)有做完的情況，有的話會(huì)等到上次作業(yè)做完莫绣，然后重新分片畴蒲，創(chuàng)建processing節(jié)點(diǎn)，再將禁用的分片項(xiàng)去除掉对室，如果失效轉(zhuǎn)移模燥，則將失效轉(zhuǎn)移的分片項(xiàng)也去除掉狞玛。在這里，會(huì)去讀取配置配置項(xiàng)sharding-total-count,job-parameter, 組裝ShardingContexts涧窒。
? jobFacade.beforeJobExecuted(shardingContexts);代碼是通知監(jiān)聽(tīng)的listener，看代碼：

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

? execute(shardingContexts,JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER);這個(gè)方法里锭亏，根據(jù)分片項(xiàng)判斷是否有分片纠吴，沒(méi)有分片項(xiàng)，結(jié)束掉調(diào)度的執(zhí)行慧瘤，如果需要向上拋出事件的戴已，拋出已完成事件，結(jié)束任務(wù)锅减。有分片任務(wù)的糖儡，去注冊(cè)作業(yè)啟動(dòng)信息，開(kāi)始執(zhí)行作業(yè)怔匣，執(zhí)行結(jié)束之后握联，將注冊(cè)信息改為結(jié)束狀態(tài)（改掉JobRegistry的狀態(tài)和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 考慮增加作業(yè)失敗的狀態(tài)每瞒，并且考慮如何處理作業(yè)失敗的整體回路
          //注冊(cè)作業(yè)的完成
            jobFacade.registerJobCompleted(shardingContexts);
            if (itemErrorMessages.isEmpty()) {
                if (shardingContexts.isAllowSendJobEvent()) {
                    jobFacade.postJobStatusTraceEvent(taskId, State.TASK_FINISHED, "");
                }
            } else {
                //是否發(fā)送jobEvent
                if (shardingContexts.isAllowSendJobEvent()) {
                    jobFacade.postJobStatusTraceEvent(taskId, State.TASK_ERROR, itemErrorMessages.toString());
                }
            }
        }
    }

? 在registerJobBegin注冊(cè)作業(yè)啟動(dòng)信息的時(shí)候金闽，首先改了JobRegistry的作業(yè)運(yùn)行狀態(tài)，JobRegistry該單例對(duì)象維護(hù)了所有job的相關(guān)信息剿骨。其次代芜，如果監(jiān)控任務(wù)執(zhí)行狀態(tài)，則創(chuàng)建作業(yè)的臨時(shí)節(jié)點(diǎn)浓利。

  /**
     * 注冊(cè)作業(yè)啟動(dòng)信息.
     * 
     * @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), "");
        }
    }

?而在作業(yè)的執(zhí)行過(guò)程中挤庇，如果作業(yè)只有一個(gè)分片，則直接去處理作業(yè)的請(qǐng)求贷掖，如果多于一個(gè)嫡秕，則使用計(jì)數(shù)器，等所有分片項(xiàng)處理完成再去統(tǒng)一返回苹威，而不是各自分片完成自己的分片任務(wù)就返回淘菩。

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();
        }
    }

作業(yè)請(qǐng)求的處理，會(huì)去調(diào)用AbstractElasticJobExecutor的process方法屠升，在這個(gè)方法里潮改，會(huì)直接調(diào)用三種基本類型的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 {
            //在這里會(huì)直接調(diào)用三種基本任務(wù)的execute方法汇在，
            //該process方法執(zhí)行的是 AbstractElasticJobExecutor
            //的process抽象方法，具體的實(shí)現(xiàn)類可看下面代碼 
            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的實(shí)現(xiàn)類
  public final class SimpleJobExecutor extends AbstractElasticJobExecutor {
    
    private final SimpleJob simpleJob;
    
    public SimpleJobExecutor(final SimpleJob simpleJob, final JobFacade jobFacade) {
        super(jobFacade);
        this.simpleJob = simpleJob;
    }
    //process方法實(shí)質(zhì)會(huì)調(diào)用三種基本任務(wù)的execute方法脏答，就是我們配置的作業(yè)的執(zhí)行方法糕殉。
    @Override
    protected void process(final ShardingContext shardingContext) {
        simpleJob.execute(shardingContext);
    }
}

jobFacade.failoverIfNecessary();作業(yè)執(zhí)行完成之后亩鬼，判斷是否需要失效轉(zhuǎn)移，再然后 jobFacade.afterJobExecuted(shardingContexts);通知監(jiān)聽(tīng)的Listenter改作業(yè)執(zhí)行完成阿蝶。

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