?在第一篇job 的類設計結構中，已經(jīng)說過job最終執(zhí)行會在quartz中執(zhí)行LiteJob該作業(yè)，LiteJob中怎樣去保證作業(yè)的執(zhí)行的？
?再看一下LiteJob的類圖：
LiteJob.png
分析下來，job的執(zhí)行過程是這張圖的樣子，比較大：
未命名文件 (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í)行過程中，首先會根據(jù)elasticJob的類型(也就是我們在使用elasticJob的過程中，配置的類型)去找到相應的執(zhí)行器，（ScriptJobExecutor，DataflowJobExecutor，DataflowJobExecutor均實現(xiàn)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);
}
?從執(zhí)行器的抽象父類構造方法看，首先會去通過jobFacade然后用configService獲取獲取job的配置，然后獲取一個執(zhí)行器服務executorService(沒有就創(chuàng)建一個executor-service-handler,不配置走默認配置)，再獲取異常處理器jobExceptionHandler(作業(yè)配置項executor-service-handler，不配置走默認配置)。
?然后看一下job的執(zhí)行過程：
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));
}
//是否有運行中的任務
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)聽對象，作業(yè)要開始執(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ān)聽器，作業(yè)執(zhí)行結束
jobFacade.afterJobExecuted(shardingContexts);
//CHECKSTYLE:OFF
} catch (final Throwable cause) {
//CHECKSTYLE:ON
jobExceptionHandler.handleException(jobName, cause);
}
}
?首先檢查環(huán)境，jobFacade.checkJobExecutionEnvironment();看一下服務器時間與注冊中心的時間誤差秒數(shù)是否在允許范圍，配置項：max-time-diff-seconds,-1為不校驗時間誤差，默認為-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（失效轉移，配置項failOver，默認配置項為false）若分片失效轉移為false，則會取判斷是否需要分片，做一系列分片邏輯，這里會去加載配置項job-sharding-strategy-class分片策略類，按照策略類分配分片策略，在這里，會去選舉主節(jié)點，然后從zk更新看是否有上次任務沒有做完的情況，有的話會等到上次作業(yè)做完，然后重新分片，創(chuàng)建processing節(jié)點，再將禁用的分片項去除掉，如果失效轉移，則將失效轉移的分片項也去除掉。在這里，會去讀取配置配置項sharding-total-count,job-parameter, 組裝ShardingContexts。
? jobFacade.beforeJobExecuted(shardingContexts);代碼是通知監(jiān)聽的listener，看代碼：
@Override
public void beforeJobExecuted(final ShardingContexts shardingContexts) {
for (ElasticJobListener each : elasticJobListeners) {
each.beforeJobExecuted(shardingContexts);
}
}
? execute(shardingContexts,JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER);這個方法里，根據(jù)分片項判斷是否有分片，沒有分片項，結束掉調(diào)度的執(zhí)行，如果需要向上拋出事件的，拋出已完成事件，結束任務。有分片任務的，去注冊作業(yè)啟動信息，開始執(zhí)行作業(yè)，執(zhí)行結束之后，將注冊信息改為結束狀態(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è)失敗的整體回路
//注冊作業(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注冊作業(yè)啟動信息的時候，首先改了JobRegistry的作業(yè)運行狀態(tài)，JobRegistry該單例對象維護了所有job的相關信息。其次，如果監(jiān)控任務執(zhí)行狀態(tài)，則創(chuàng)建作業(yè)的臨時節(jié)點。
/**
* 注冊作業(yè)啟動信息.
*
* @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í)行過程中，如果作業(yè)只有一個分片，則直接去處理作業(yè)的請求，如果多于一個，則使用計數(shù)器，等所有分片項處理完成再去統(tǒng)一返回，而不是各自分片完成自己的分片任務就返回。
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è)請求的處理，會去調(diào)用AbstractElasticJobExecutor的process方法，在這個方法里，會直接調(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 {
//在這里會直接調(diào)用三種基本任務的execute方法，
//該process方法執(zhí)行的是 AbstractElasticJobExecutor
//的process抽象方法，具體的實現(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的實現(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方法實質(zhì)會調(diào)用三種基本任務的execute方法，就是我們配置的作業(yè)的執(zhí)行方法。
@Override
protected void process(final ShardingContext shardingContext) {
simpleJob.execute(shardingContext);
}
}
jobFacade.failoverIfNecessary();作業(yè)執(zhí)行完成之后，判斷是否需要失效轉移，再然后 jobFacade.afterJobExecuted(shardingContexts);通知監(jiān)聽的Listenter改作業(yè)執(zhí)行完成。
@Override
public void afterJobExecuted(final ShardingContexts shardingContexts) {
for (ElasticJobListener each : elasticJobListeners) {
each.afterJobExecuted(shardingContexts);
}
}
