5. xxljob源码分析调度触发原理（一）
上篇分析了XxlJobScheduler的主要流程和各个流程具体的逻辑内容，但是并没有深入到核心细节原理。本篇接着JobTriggerPoolHelper类的后半部分接着探索。看一看为什么作者需要分快触发和慢触发。具体触发的逻辑和原理又是什么。
JobTriggerPoolHelper.trigger
public void addTrigger(final int jobId, final TriggerTypeEnum triggerType, final int failRetryCount, final String executorShardingParam, final String executorParam, final String addressList) { // 选择一个触发的线程池，默认是快触发 ThreadPoolExecutor triggerPool_ = fastTriggerPool; AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId); if (jobTimeoutCount!=null && jobTimeoutCount.get() > 10) { // 1分钟内超时了10次就放入到慢触发线程池 triggerPool_ = slowTriggerPool; } // 线程池触发调用 triggerPool_.execute(new Runnable() { @Override public void run() { long start = System.currentTimeMillis(); try { // 触发 XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList); } catch (Exception e) { logger.error(e.getMessage(), e); } finally { // 1分钟清理一次超时次数的map long minTim_now = System.currentTimeMillis()/60000; if (minTim != minTim_now) { minTim = minTim_now; jobTimeoutCountMap.clear(); } // 计算花费时间 long cost = System.currentTimeMillis()-start; if (cost > 500) { // 超过500ms了就算是超时一次 AtomicInteger timeoutCount = jobTimeoutCountMap.putIfAbsent(jobId, new AtomicInteger(1)); if (timeoutCount != null) { timeoutCount.incrementAndGet(); } } } } }); }
上述代码可以看到默认使用的线程池是快速触发线程池，而当触发的超时次数超过了1分钟10次的话，就放入另外一个慢触发线程池。这样做的目的应该是怕慢触发的任务影响了快速触发的任务。因为线程池的线程创建的个数是有限的，超过之后，就会放入任务队列等待执行。如果任务量大，执行的速度又慢，实际会影响其他任务的处理。分成两个可以更好的区分任务的执行快慢，分开触发，影响小。
这里判断是否超时的条件是任务的执行时间是否大于了500ms，超过500ms就进行累加次数。超时的次数会存入一个单独的map，其key为jobId，并且每分钟做一次清理操作。
继续看下XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList)方法
public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam, String addressList) { // 获取job信息 XxlJobInfo jobInfo = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(jobId); if (jobInfo == null) { logger.warn(">>>>>>>>>>>> trigger fail, jobId invalid，jobId={}", jobId); return; } if (executorParam != null) { // 设置参数 jobInfo.setExecutorParam(executorParam); } int finalFailRetryCount = failRetryCount>=0?failRetryCount:jobInfo.getExecutorFailRetryCount(); XxlJobGroup group = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().load(jobInfo.getJobGroup()); // 设置地址 if (addressList!=null && addressList.trim().length()>0) { group.setAddressType(1); group.setAddressList(addressList.trim()); } // 设置分片参数 int[] shardingParam = null; if (executorShardingParam!=null){ String[] shardingArr = executorShardingParam.split("/"); if (shardingArr.length==2 && isNumeric(shardingArr[0]) && isNumeric(shardingArr[1])) { shardingParam = new int[2]; shardingParam[0] = Integer.valueOf(shardingArr[0]); shardingParam[1] = Integer.valueOf(shardingArr[1]); } } if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null) && group.getRegistryList()!=null && !group.getRegistryList().isEmpty() && shardingParam==null) { // 分片广播的处理，也就是根据注册的执行器，每隔执行器执行自己分片的那一部分，传递了总分片和当前自己处理的分片 for (int i = 0; i < group.getRegistryList().size(); i++) { processTrigger(group, jobInfo, finalFailRetryCount, triggerType, i, group.getRegistryList().size()); } } else { // 非广播分片直接调用处理 if (shardingParam == null) { shardingParam = new int[]{0, 1}; } processTrigger(group, jobInfo, finalFailRetryCount, triggerType, shardingParam[0], shardingParam[1]); } }
在触发的时候，首先是对参数信息进行了处理，然后判断了注册了执行器的数量和配置的策略，如果支持广播分片，还需要进行分片的处理，每个执行器执行自己负责的那个分片，否则直接执行，分片也就是默认的单片。这个方法处理完分片的信息后，就调用了processTrigger。
private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){ // 获取参数 ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION); // block strategy ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); // route strategy String shardingParam = (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==executorRouteStrategyEnum)?String.valueOf(index).concat("/").concat(String.valueOf(total)):null; // 1、保存日志id XxlJobLog jobLog = new XxlJobLog(); jobLog.setJobGroup(jobInfo.getJobGroup()); jobLog.setJobId(jobInfo.getId()); jobLog.setTriggerTime(new Date()); XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().save(jobLog); logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId()); // 2、初始化触发参数 TriggerParam triggerParam = new TriggerParam(); triggerParam.setJobId(jobInfo.getId()); triggerParam.setExecutorHandler(jobInfo.getExecutorHandler()); triggerParam.setExecutorParams(jobInfo.getExecutorParam()); triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy()); triggerParam.setExecutorTimeout(jobInfo.getExecutorTimeout()); triggerParam.setLogId(jobLog.getId()); triggerParam.setLogDateTime(jobLog.getTriggerTime().getTime()); triggerParam.setGlueType(jobInfo.getGlueType()); triggerParam.setGlueSource(jobInfo.getGlueSource()); triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime()); triggerParam.setBroadcastIndex(index); triggerParam.setBroadcastTotal(total); // 3、初始化调用执行的地址 String address = null; ReturnT routeAddressResult = null; if (group.getRegistryList()!=null && !group.getRegistryList().isEmpty()) { if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) { if (index >>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +">>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId()); }
在processTrigger方法中，第一步是初始化了日志的信息，并记录了一条初始化的日志。接着是初始化触发的参数。这些参数信息都是需要传递给执行器进行调用的必要信息，包括最重要的执行Handler。然后是需要根据分片的信息拿到调用执行器的地址。分片的概念实际是将任务分解，然后进行分布式的任务处理。因为每个执行器执行属于自己的一部分任务，互不干扰，这样就可以实行多台机器并行的执行，加快任务的处理。后续就是触发执行和触发后的日志信息的保存。毕竟不能保证每个任务都能成功。当任务触发失败后，需要详细的记录下日志信息，这样才方便后续问题的排查。
但是上述代码任然没触及到核心具体是怎么触发的。继续深入triggerResult = runExecutor(triggerParam, address);
public static ReturnT runExecutor(TriggerParam triggerParam, String address){ ReturnT runResult = null; try { // 获取到触发执行器 ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address); // 触发执行 runResult = executorBiz.run(triggerParam); } catch (Exception e) { logger.error(">>>>>>>>>>> xxl-job trigger error, please check if the executor[{}] is running.", address, e); runResult = new ReturnT(ReturnT.FAIL_CODE, ThrowableUtil.toString(e)); } StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + "："); runResultSB.append("address：").append(address); runResultSB.append("code：").append(runResult.getCode()); runResultSB.append("msg：").append(runResult.getMsg()); runResult.setMsg(runResultSB.toString()); return runResult; }
这里获取到执行器后就可以触发执行了。但是这个执行器有两个实现，具体是哪一个呢？
public static ExecutorBiz getExecutorBiz(String address) throws Exception { // valid if (address==null || address.trim().length()==0) { return null; } // load-cache address = address.trim(); ExecutorBiz executorBiz = executorBizRepository.get(address); if (executorBiz != null) { return executorBiz; } // set-cache executorBiz = new ExecutorBizClient(address, XxlJobAdminConfig.getAdminConfig().getAccessToken()); executorBizRepository.put(address, executorBiz); return executorBiz; }
如果根据远程地址获取不到执行器，就需要创建一个，也就是ExecutorBizClient这个执行器。最后查看下ExecutorBizClient的run方法。
public ReturnT run(TriggerParam triggerParam) { // 发送post请求到各个执行器调度处理 return XxlJobRemotingUtil.postBody(addressUrl + "run", accessToken, timeout, triggerParam, String.class); }
分析到这里就很清晰了，run方法调用注册到调度中心的执行器的一个post请求，让任务执行器去执行。任务执行器接受到请求后，根据参数就可以找到对应的Handler进行逻辑处理了。
总结
本篇分析了作者创建快触发和慢触发线程池的理由。也重点分析trigger方法。这里面并没有很难很深的东西。其实很多源码也并不是太难。跟着作者的思路走，都能理解。关键是在看源码的过程中，要去思考作者的设计。如此设计的好处在哪里。在上面的代码中，作者就考虑了触发任务的执行快慢，对触发任务记录了日志，方便后续的问题排查。还设计了分片的思想，方便我们在多个处理器上进行分布式的执行任务。如果是我们自己设计，自己写这个任务调度是不是能考虑的如此全面呢？说到这里，实际上你就明白差距在哪里了，应该学习什么内容了。
不过本篇没有介绍调度触发的时机，也就是为什么它能够精准的调度，这个我们下一篇继续探讨。
