透过源码看xxl-job
（注：本文基于xxl-job最新版v2.0.2, quartz版本为 v2.3.1。 以下提到的调度中心均指xxl-job-admin项目）
上回说到，xxl-job是一个中心化的设计方案，分为了调度中心和 执行器两部分。其本质上仍然是对quartz的封装。那么，我们就分别通过“调度中心” 和 “执行器” 来看看它是怎么运作的。
调度中心
初始化
由于是spring boot应用，因此先从配置看起。
XxlJobDynamicSchedulerConfig
相关的初始化是在XxlJobDynamicSchedulerConfig中完成的，下面我们来看XxlJobDynamicSchedulerConfig源码。
@Configuration
public class XxlJobDynamicSchedulerConfig { @Bean public SchedulerFactoryBean getSchedulerFactoryBean(DataSource dataSource){ SchedulerFactoryBean schedulerFactory = new SchedulerFactoryBean(); schedulerFactory.setDataSource(dataSource); schedulerFactory.setAutoStartup(true); // 自动启动 schedulerFactory.setStartupDelay(20); // 延时启动，应用启动成功后在启动 schedulerFactory.setOverwriteExistingJobs(true); // 覆盖DB中JOB：true、以数据库中已经存在的为准：false schedulerFactory.setApplicationContextSchedulerContextKey("applicationContext"); schedulerFactory.setConfigLocation(new ClassPathResource("quartz.properties")); return schedulerFactory; } @Bean(initMethod = "start", destroyMethod = "destroy") public XxlJobDynamicScheduler getXxlJobDynamicScheduler(SchedulerFactoryBean schedulerFactory){ Scheduler scheduler = schedulerFactory.getScheduler(); XxlJobDynamicScheduler xxlJobDynamicScheduler = new XxlJobDynamicScheduler(); xxlJobDynamicScheduler.setScheduler(scheduler); return xxlJobDynamicScheduler; } }
由上可知 XxlJobDynamicSchedulerConfig 主要是创建了 SchedulerFactoryBean 对XxlJobDynamicScheduler 对象。SchedulerFactoryBean创建调度器（Scheduler, 没错，它就是quartz中的Scheduler对象）， XxlJobDynamicScheduler持有对Scheduler对象的引用。
那么，SchedulerFactoryBean 是如何创建 Scheduler的呢，接下来，我们再看看SchedulerFactoryBean 。
SchedulerFactoryBean
SchedulerFactoryBean实现了InitializingBean， 其主要初始化流程在 afterPropertiesSet 方法中。
@Override public void afterPropertiesSet() throws Exception { if (this.dataSource == null && this.nonTransactionalDataSource != null) { this.dataSource = this.nonTransactionalDataSource; } if (this.applicationContext != null && this.resourceLoader == null) { this.resourceLoader = this.applicationContext; } // 初始化scheduler this.scheduler = prepareScheduler(prepareSchedulerFactory()); try { registerListeners(); registerJobsAndTriggers(); } catch (Exception ex) { try { this.scheduler.shutdown(true); } catch (Exception ex2) { logger.debug("Scheduler shutdown exception after registration failure", ex2); } throw ex; } }
以上，先通过prepareSchedulerFactory方法创建ScheduleFactory对象（quartz），再通过prepareScheduler方法创建Scheduler对象。
private SchedulerFactory prepareSchedulerFactory() throws SchedulerException, IOException { SchedulerFactory schedulerFactory = this.schedulerFactory; if (schedulerFactory == null) { //默认为StdSchedulerFactory schedulerFactory = BeanUtils.instantiateClass(this.schedulerFactoryClass); if (schedulerFactory instanceof StdSchedulerFactory) { //解析处理配置 initSchedulerFactory((StdSchedulerFactory) schedulerFactory); } else if (this.configLocation != null || this.quartzProperties != null || this.taskExecutor != null || this.dataSource != null) { throw new IllegalArgumentException("StdSchedulerFactory required for applying Quartz properties: " + schedulerFactory); } } return schedulerFactory; }
接下来，我们看看prepareScheduler方法是怎么创建scheduler对象的。
protected Scheduler createScheduler(SchedulerFactory schedulerFactory, String schedulerName) throws SchedulerException { ......... //已省略部分我们本次不用关心的代码 try { SchedulerRepository repository = SchedulerRepository.getInstance(); synchronized (repository) { Scheduler existingScheduler = (schedulerName != null ? repository.lookup(schedulerName) : null); //通过schedulerFactory创建scheduler, 重点关注。前往quartz中一探究竟 Scheduler newScheduler = schedulerFactory.getScheduler(); if (newScheduler == existingScheduler) { throw new IllegalStateException("Active Scheduler of name '" + schedulerName + "' already registered " + "in Quartz SchedulerRepository. Cannot create a new Spring-managed Scheduler of the same name!"); } if (!this.exposeSchedulerInRepository) { SchedulerRepository.getInstance().remove(newScheduler.getSchedulerName()); } return newScheduler; } } finally { if (overrideClassLoader) { // Reset original thread context ClassLoader. currentThread.setContextClassLoader(threadContextClassLoader); } } }
ok， 接着前往quartz中一探究竟。
StdSchedulerFactory
public Scheduler getScheduler() throws SchedulerException { if (cfg == null) { initialize(); } SchedulerRepository schedRep = SchedulerRepository.getInstance(); Scheduler sched = schedRep.lookup(getSchedulerName()); //如果存在该对象，则直接返回 if (sched != null) { if (sched.isShutdown()) { schedRep.remove(getSchedulerName()); } else { return sched; } } //重点关注 sched = instantiate(); return sched; }
下面就重点看看instantiate方法。
private Scheduler instantiate() throws SchedulerException { .... QuartzSchedulerResources rsrcs = new QuartzSchedulerResources(); rsrcs.setName(schedName); rsrcs.setThreadName(threadName); rsrcs.setInstanceId(schedInstId); rsrcs.setJobRunShellFactory(jrsf); rsrcs.setMakeSchedulerThreadDaemon(makeSchedulerThreadDaemon); rsrcs.setThreadsInheritInitializersClassLoadContext(threadsInheritInitalizersClassLoader); rsrcs.setBatchTimeWindow(batchTimeWindow); rsrcs.setMaxBatchSize(maxBatchSize); rsrcs.setInterruptJobsOnShutdown(interruptJobsOnShutdown); rsrcs.setInterruptJobsOnShutdownWithWait(interruptJobsOnShutdownWithWait); rsrcs.setJMXExport(jmxExport); rsrcs.setJMXObjectName(jmxObjectName); SchedulerDetailsSetter.setDetails(tp, schedName, schedInstId); rsrcs.setThreadExecutor(threadExecutor); threadExecutor.initialize(); rsrcs.setThreadPool(tp); if(tp instanceof SimpleThreadPool) { if(threadsInheritInitalizersClassLoader) ((SimpleThreadPool)tp).setThreadsInheritContextClassLoaderOfInitializingThread(threadsInheritInitalizersClassLoader); } tp.initialize(); tpInited = true; rsrcs.setJobStore(js); // add plugins for (int i = 0; i < plugins.length; i++) { rsrcs.addSchedulerPlugin(plugins[i]); } //创建QuartzScheduler对象，重点关注，此为Quartz核心部分 qs = new QuartzScheduler(rsrcs, idleWaitTime, dbFailureRetry); qsInited = true; // 创建Scheduler对象，QuartzScheduler并未直接实现Scheduler接口，而是作为了Scheduler的委托者 Scheduler scheduler = instantiate(rsrcs, qs); ... } protected Scheduler instantiate(QuartzSchedulerResources rsrcs, QuartzScheduler qs) { Scheduler scheduler = new StdScheduler(qs); return scheduler; }
以上代码是经过大刀阔斧砍掉过的，原代码十分长，通篇下来主要是根据配置去创建一系列的对象，所有的对象最终都将被以上代码中的 QuartzSchedulerResources 对象所持有，这些对象共同协作才能最终组装出Quartz这台"机器"， 通过以上代码也可大致窥探出创建了哪些对象实例，这些对象实例的创建大多都可通过quartz.properties进行配置。
其中，我们更应该关注的是 QuartzScheduler 对象的创建，它实则为Quartz的心脏。
QuartzScheduler
public QuartzScheduler(QuartzSchedulerResources resources, long idleWaitTime, @Deprecated long dbRetryInterval) throws SchedulerException { this.resources = resources; if (resources.getJobStore() instanceof JobListener) { addInternalJobListener((JobListener)resources.getJobStore()); } //创建QuartzSchedulerThread对象，重点关注，此线程负责任务调度 this.schedThread = new QuartzSchedulerThread(this, resources); ThreadExecutor schedThreadExecutor = resources.getThreadExecutor(); //DefaultThreadExecutor对象，该方法的作用是启动schedThread线程 schedThreadExecutor.execute(this.schedThread); if (idleWaitTime > 0) { this.schedThread.setIdleWaitTime(idleWaitTime); } jobMgr = new ExecutingJobsManager(); addInternalJobListener(jobMgr); errLogger = new ErrorLogger(); addInternalSchedulerListener(errLogger); signaler = new SchedulerSignalerImpl(this, this.schedThread); getLog().info("Quartz Scheduler v." + getVersion() + " created."); }
以上代码，主要是创建了QuartzSchedulerThread对象，然后通过DefaultThreadExecutor进行启动。
QuartzSchedulerThread
QuartzSchedulerThread实现自Thread，我们接下来就看看其核心代码。
@Override public void run() { int acquiresFailed = 0; //是否结束循环\调度 while (!halted.get()) { try { synchronized (sigLock) { //如果是暂停状态，则在此阻塞，直到外部更改状态 while (paused && !halted.get()) { try { sigLock.wait(1000L); } catch (InterruptedException ignore) { } acquiresFailed = 0; } if (halted.get()) { break; } } ...... //获取可用线程数量 int availThreadCount =qsRsrcs.getThreadPool().blockForAvailableThreads(); if(availThreadCount > 0) { List<OperableTrigger> triggers; long now = System.currentTimeMillis(); clearSignaledSchedulingChange(); //从DB中取出一批即将要执行的Trigger(触发器), DB中该数据状态也会同步进行修改 try { triggers = qsRsrcs.getJobStore().acquireNextTriggers( now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow()); acquiresFailed = 0; if (log.isDebugEnabled()) log.debug("batch acquisition of " + (triggers == null ? 0 : triggers.size()) + " triggers"); } catch (JobPersistenceException jpe) { if (acquiresFailed == 0) { qs.notifySchedulerListenersError("An error occurred while scanning for the next triggers to fire.",jpe); } if (acquiresFailed < Integer.MAX_VALUE) acquiresFailed++; continue; } catch (RuntimeException e) { if (acquiresFailed == 0) { getLog().error("quartzSchedulerThreadLoop: RuntimeException " +e.getMessage(), e); } if (acquiresFailed < Integer.MAX_VALUE) acquiresFailed++; continue; } if (triggers != null && !triggers.isEmpty()) { ...... List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>(); boolean goAhead = true; synchronized(sigLock) { goAhead = !halted.get(); } if(goAhead) { //取出触发器对应的任务，同步修改相关DB中的记录状态，并调整下次执行时间 try { List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers); if(res != null) bndles = res; } catch (SchedulerException se) { qs.notifySchedulerListenersError("An error occurred while firing triggers '"+ triggers + "'", se); for (int i = 0; i < triggers.size(); i++) { qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i)); } continue; } } //真正执行的方法，包装为JobRunShell， 并从线程池中获取线程进行执行 for (int i = 0; i < bndles.size(); i++) { TriggerFiredResult result = bndles.get(i); TriggerFiredBundle bndle = result.getTriggerFiredBundle(); Exception exception = result.getException(); if (exception instanceof RuntimeException) { getLog().error("RuntimeException while firing trigger " + triggers.get(i), exception); qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i)); continue; } if (bndle == null) { qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i)); continue; } JobRunShell shell = null; try { shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle); shell.initialize(qs); } catch (SchedulerException se) { qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR); continue; } if (qsRsrcs.getThreadPool().runInThread(shell) == false) { getLog().error("ThreadPool.runInThread() return false!"); qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR); } } continue; // while (!halted) } } else { continue; // while (!halted) } ...... }
以上代码同样进行过精简，该方法为quartz的核心调度流程。由于内部业务较为复杂，只在代码上加了简单的注释，不过主要流程就是 从DB中获取Trigger触发器和Job（任务）, 同时通过更新DB数据状态来防止集群下的“争抢”，通过线程的wait和notify机制来协同线程调度，最终从线程池中获取线程来执行我们的任务。
ok， 到此，quartz这颗小心脏就已经跳动起来了。
那么，到此结束？
No, 一切才刚刚开始！ 说好的xxl-job呢？
XxlJobDynamicScheduler
回到XxlJobDynamicSchedulerConfig，我们发现在初始化XxlJobDynamicScheduler对象后，会调用其start方法。那么，我们进入其start方法一探究竟。
public void start() throws Exception { // valid Assert.notNull(scheduler, "quartz scheduler is null"); // 国际化 initI18n(); // 启动维护执行器注册信息守护线程 JobRegistryMonitorHelper.getInstance().start(); // 启动执行失败的任务扫描守护线程 JobFailMonitorHelper.getInstance().start(); // 初始化RPC （接收执行器注册和回调等）， 在分析执行器的时候再来看，本次不看 initRpcProvider(); logger.info(">>>>>>>>> init xxl-job admin success."); }
当执行器自动注册后，调度中心是如何去维护它的呢？ 答案就在 JobRegistryMonitorHelper 线程里。
JobRegistryMonitorHelper
public void start(){ registryThread = new Thread(new Runnable() { @Override public void run() { while (!toStop) { try { // 从XXL_JOB_QRTZ_TRIGGER_GROUP表中获取自动注册类型的执行器 List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().findByAddressType(0); if (groupList!=null && !groupList.isEmpty()) { //注册信息记录在XXL_JOB_QRTZ_TRIGGER_REGISTRY表，删除90秒没心跳机器 XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(RegistryConfig.DEAD_TIMEOUT); HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>(); //XXL_JOB_QRTZ_TRIGGER_REGISTRY表获取存活的机器 List<XxlJobRegistry> list =XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findAll(RegistryConfig.DEAD_TIMEOUT); //appname 相同的形成集群 if (list != null) { for (XxlJobRegistry item: list) { if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) { String appName = item.getRegistryKey(); List<String> registryList = appAddressMap.get(appName); if (registryList == null) { registryList = new ArrayList<String>(); } if (!registryList.contains(item.getRegistryValue())) { registryList.add(item.getRegistryValue()); } appAddressMap.put(appName, registryList); } } } // 维护集群地址（XXL_JOB_QRTZ_TRIGGER_GROUP表，地址逗号分隔） for (XxlJobGroup group: groupList) { List<String> registryList = appAddressMap.get(group.getAppName()); String addressListStr = null; if (registryList!=null && !registryList.isEmpty()) { Collections.sort(registryList); addressListStr = ""; for (String item:registryList) { addressListStr += item + ","; } addressListStr = addressListStr.substring(0, addressListStr.length()-1); } group.setAddressList(addressListStr); XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group); } } } catch (Exception e) { ...... } try { TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT); } catch (InterruptedException e) { ...... } } logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop"); } }); registryThread.setDaemon(true); registryThread.setName("xxl-job, admin JobRegistryMonitorHelper"); registryThread.start(); }
关于注册信息的维护比较简单，就是定时检查有没心跳，心跳体现在DB中（通过每次更新DB记录时间，来表示存活）。一定时间窗口内（默认90秒）没更新心跳的，就认为已经dead, 直接剔除，然后维护当前存活机器的地址。
JobFailMonitorHelper
当任务执行失败时，我们需要收到邮件报警。甚至有时候我们需要任务进行自动重试，那么，xxl-job是如何实现的呢？ 答案就在 JobFailMonitorHelper 中。
public void start(){ monitorThread = new Thread(new Runnable() { @Override public void run() { // monitor while (!toStop) { try { //XXL_JOB_QRTZ_TRIGGER_LOG表中记录的是任务执行 //从XXL_JOB_QRTZ_TRIGGER_LOG表中取出执行失败的记录 List<Integer> failLogIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().findFailJobLogIds(1000); if (failLogIds!=null && !failLogIds.isEmpty()) { for (int failLogId: failLogIds) { //锁定日志记录 int lockRet = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, 0, -1); if (lockRet < 1) { continue; } XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(failLogId); //XXL_JOB_QRTZ_TRIGGER_INFO表中获取任务详情 XxlJobInfo info = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(log.getJobId()); // 没达到最大重试次数，则进行重试，日志中记录的就是剩余的重试次数 if (log.getExecutorFailRetryCount() > 0) { //发起重试（触发流程参考后面章节） JobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam(), null); ....... //更新日志 XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(log); } // 失败任务报警 // 0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败 int newAlarmStatus = 0; if (info!=null && info.getAlarmEmail()!=null && info.getAlarmEmail().trim().length()>0) { boolean alarmResult = true; try { alarmResult = failAlarm(info, log); } catch (Exception e) { alarmResult = false; logger.error(e.getMessage(), e); } newAlarmStatus = alarmResult?2:3; } else { newAlarmStatus = 1; } //更新报警状态 XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, -1, newAlarmStatus); } } TimeUnit.SECONDS.sleep(10); } catch (Exception e) { ...... } } logger.info(">>>>>>>>>>> xxl-job, job fail monitor thread stop"); } }); monitorThread.setDaemon(true); monitorThread.setName("xxl-job, admin JobFailMonitorHelper"); monitorThread.start(); }
至此，调度中心我们关心的主要流程就已经初始化完毕。现在，我们大致清楚了xxl-job初始化流程，调度中心对于我们而言，其核心功能无非对任务进行增删改查的管理以及触发和停止，增删改查还好，其实质就是对于DB的CRUD操作，但是触发调度和停止任务是怎么做的呢？ 由于xxl-job是调度中心和执行器分离的，所以，上述问题换句话来说就是两者间是如何通信的。
答案就是RPC， 接下来，我们通过调度一个任务，来看看其执行流程。
执行流程
打开调度中心页面，在任务操作栏点击 “启动” 按钮，会发现其请求路径为 “/jobinfo/start”， 都到这一步了，学WEB是不是该秒懂，马上前往 /jobinfo/start。
@Controller @RequestMapping("/jobinfo") public class JobInfoController { ...... @RequestMapping("/start") @ResponseBody public ReturnT<String> start(int id) { return xxlJobService.start(id); } ...... }
@Service public class XxlJobServiceImpl implements XxlJobService { ...... @Override public ReturnT<String> start(int id) { //XXL_JOB_QRTZ_TRIGGER_INFO表获取任务信息 XxlJobInfo xxlJobInfo = xxlJobInfoDao.loadById(id); String name = String.valueOf(xxlJobInfo.getId()); //获取cron表达式 String cronExpression = xxlJobInfo.getJobCron(); try { boolean ret = XxlJobDynamicScheduler.addJob(name, cronExpression); return ret?ReturnT.SUCCESS:ReturnT.FAIL; } catch (SchedulerException e) { logger.error(e.getMessage(), e); return ReturnT.FAIL; } } }
public class XxlJobDynamicScheduler { public static boolean addJob(String jobName, String cronExpression) throws SchedulerException { ...... CronTrigger cronTrigger = TriggerBuilder.newTrigger(). withIdentity(triggerKey).withSchedule(cronScheduleBuilder).build(); // 任务最终将转换为RemoteHttpJobBean Class<? extends Job> jobClass_ = RemoteHttpJobBean.class; JobDetail jobDetail = JobBuilder.newJob(jobClass_).withIdentity(jobKey).build(); // 通过quartz的scheduler (StdScheduler)调度任务 Date date = scheduler.scheduleJob(jobDetail, cronTrigger); return true; } }
public class StdScheduler { ... private QuartzScheduler sched; ... public Date scheduleJob(JobDetail jobDetail, Trigger trigger) throws SchedulerException { //来到了我们之前说的quartz的心脏部分QuartzScheduler return sched.scheduleJob(jobDetail, trigger); } }
public class QuartzScheduler implements RemotableQuartzScheduler { ...... private SchedulerSignaler signaler; ...... public Date scheduleJob(JobDetail jobDetail, Trigger trigger) throws SchedulerException { ...... //唤醒线程 notifySchedulerThread(trigger.getNextFireTime().getTime()); ...... return ft; } protected void notifySchedulerThread(long candidateNewNextFireTime) { if (isSignalOnSchedulingChange()) { //通过SchedulerSignalerImpl会调用到signalSchedulingChange方法 //SchedulerSignalerImpl.schedThread.signalSchedulingChange(candidateNewNextFireTime); signaler.signalSchedulingChange(candidateNewNextFireTime); } } public void signalSchedulingChange(long candidateNewNextFireTime) { synchronized(sigLock) { signaled = true; signaledNextFireTime = candidateNewNextFireTime; //唤醒线程 sigLock.notifyAll(); } } }
至此，一切又回到了我们之前介绍过的 QuartzScheduler。刚刚，提到我们的任务类型最终会被注册为RemoteHttpJobBean，这发生在哪一步？ 其实就发生在 JobRunShell （之前提到所有任务都会被包装为JobRunShell 对象，然后在线程池中获取线程执行）中的initialize方法。
public class JobRunShell extends SchedulerListenerSupport implements Runnable { ...... public void initialize(QuartzScheduler sched) throws SchedulerException { this.qs = sched; Job job = null; JobDetail jobDetail = firedTriggerBundle.getJobDetail(); try { //最终通过jobdetail的jobClass创建实例， //这个jobClass正是我们上面设置的 RemoteHttpJobBean job = sched.getJobFactory().newJob(firedTriggerBundle, scheduler); } catch (SchedulerException se) { sched.notifySchedulerListenersError("An error occured instantiating job to be executed. job= '"+ jobDetail.getKey() + "'", se); throw se; } catch (Throwable ncdfe) { // such as NoClassDefFoundError SchedulerException se = new SchedulerException("Problem instantiating class '"+ jobDetail.getJobClass().getName() + "' - ", ncdfe); sched.notifySchedulerListenersError("An error occured instantiating job to be executed. job= '"+ jobDetail.getKey() + "'", se); throw se; } this.jec = new JobExecutionContextImpl(scheduler, firedTriggerBundle, job); } ...... //启动 public void run() { qs.addInternalSchedulerListener(this); try { OperableTrigger trigger = (OperableTrigger) jec.getTrigger(); JobDetail jobDetail = jec.getJobDetail(); do { JobExecutionException jobExEx = null; Job job = jec.getJobInstance(); ...... try { log.debug("Calling execute on job " + jobDetail.getKey()); //执行任务，调用RemoteHttpJobBean的executeInternal方法 job.execute(jec); endTime = System.currentTimeMillis(); } catch (JobExecutionException jee) { ...... } catch (Throwable e) { ...... } jec.setJobRunTime(endTime - startTime); ...... qs.notifyJobStoreJobComplete(trigger, jobDetail, instCode); break; } while (true); } finally { qs.removeInternalSchedulerListener(this); } } }
以上，JobRunShell线程启动时，最终会调用RemoteHttpJobBean的executeInternal方法。
public class RemoteHttpJobBean extends QuartzJobBean { private static Logger logger = LoggerFactory.getLogger(RemoteHttpJobBean.class); @Override protected void executeInternal(JobExecutionContext context) throws JobExecutionException { // load jobId JobKey jobKey = context.getTrigger().getJobKey(); Integer jobId = Integer.valueOf(jobKey.getName()); // 实际调用JobTriggerPoolHelper.addTrigger方法，看下面代码 JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null); } }
public class JobTriggerPoolHelper { ...... private static JobTriggerPoolHelper helper = new JobTriggerPoolHelper(); ...... public static void trigger(int jobId, TriggerTypeEnum triggerType,int failRetryCount,String executorShardingParam,String executorParam) { helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam); } public void addTrigger(final int jobId, final TriggerTypeEnum triggerType,final int failRetryCount, final String executorShardingParam, final String executorParam) { // 根据任务执行时间进行了线程池隔离，分快慢两个线程池，默认为快线程池 ThreadPoolExecutor triggerPool_ = fastTriggerPool; AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId); //在一定窗口期内(默认1分钟)达到条件（时间大于500毫秒10次）则进入慢线程池 if (jobTimeoutCount!=null && jobTimeoutCount.get() > 10) { triggerPool_ = slowTriggerPool; } // 通过线程池执行 triggerPool_.execute(new Runnable() { @Override public void run() { long start = System.currentTimeMillis(); try { // 重点关注，到此时才是真正触发执行 XxlJobTrigger.trigger(jobId, triggerType, failRetryCount,executorShardingParam, executorParam); } catch (Exception e) { logger.error(e.getMessage(), e); } finally { // 时间窗口为1分钟，超过就清空，进入下一个周期 long minTim_now = System.currentTimeMillis()/60000; if (minTim != minTim_now) { minTim = minTim_now; jobTimeoutCountMap.clear(); } // 每超过500毫秒就记录超时一次 long cost = System.currentTimeMillis()-start; if (cost > 500) { AtomicInteger timeoutCount = jobTimeoutCountMap.put(jobId, new AtomicInteger(1)); if (timeoutCount != null) { timeoutCount.incrementAndGet(); } } } } }); } }
以上代码，我们可以清楚看到xxl-job对于线程池隔离的处理规则，其实对于我们在设计同类问题的时候还是具有一定的参考价值。当然，本段代码最值得我们关注的还是其真正调用了XxlJobTrigger的trigger方法，这才是最终真正触发任务执行的。作了这么多准备，似乎好戏才真正开始。
public class XxlJobTrigger { ...... public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam) { // XXL_JOB_QRTZ_TRIGGER_INFO表获取任务信息 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(); //XXL_JOB_QRTZ_TRIGGER_GROUP表获取执行器相关信息 XxlJobGroup group = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().load(jobInfo.getJobGroup()); // 如果有分片，就进行分片处理 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) { //如果是SHARDING_BROADCAST（分片广播策略），则对应所有执行器都将被触发 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]); } } ...... 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); //路由策略 ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); //分片参数 String shardingParam = (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==executorRouteStrategyEnum)?String.valueOf(index).concat("/").concat(String.valueOf(total)):null; // 记录日志 XxlJobLog jobLog = new XxlJobLog(); ...... XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().save(jobLog); // 组装TriggerParam参数 TriggerParam triggerParam = new TriggerParam(); ...... // 获取相应的执行器地址 String address = null; ReturnT<String> routeAddressResult = null; if (group.getRegistryList()!=null && !group.getRegistryList().isEmpty()) { if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) { //如果是分片广播，就根据当前分片序号，取出执行器地址 if (index < group.getRegistryList().size()) { address = group.getRegistryList().get(index); } else { address = group.getRegistryList().get(0); } } else { //根据路由策略获取相应执行器地址 //一些列路由策略继承自ExecutorRouter routeAddressResult = executorRouteStrategyEnum.getRouter().route(triggerParam, group.getRegistryList()); if (routeAddressResult.getCode() == ReturnT.SUCCESS_CODE) { address = routeAddressResult.getContent(); } } } else { routeAddressResult = new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobconf_trigger_address_empty")); } //执行 ReturnT<String> triggerResult = null; if (address != null) { //经过一系列组装参数，路由选址后，最终开始执行，该方法在下面，重点关注 triggerResult = runExecutor(triggerParam, address); } else { triggerResult = new ReturnT<String>(ReturnT.FAIL_CODE, null); } ...... //更新日志 XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(jobLog); logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId()); } /** * 最终执行的地方 * @param triggerParam * @param address * @return */ public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){ ReturnT<String> runResult = null; try { //此处获取的为代理对象（注意） ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(address); //真正执行的为代理对象 runResult = executorBiz.run(triggerParam); } catch (Exception e) { ...... } ...... return runResult; } }
到此，我们离真相只差最后一步了。上面获取ExecutorBiz对象，然后通过ExecutorBiz进行最终执行，特别需要注意的是获取到的ExecutorBiz是个代理对象。如果没打开XxlJobDynamicScheduler.getExecutorBiz进行查看，直接点run, 你会觉得你的打开方式没对。
那么，最后，我们就来通过这个代理对象解开最后谜题吧。
public final class XxlJobDynamicScheduler { ...... private static ConcurrentHashMap<String, ExecutorBiz> executorBizRepository = new ConcurrentHashMap<String, ExecutorBiz>(); /** * 获取ExecutorBiz代理对象 * @param address * @return * @throws Exception */ public static ExecutorBiz getExecutorBiz(String address) throws Exception { if (address==null || address.trim().length()==0) { return null; } // 从缓存中获取 address = address.trim(); ExecutorBiz executorBiz = executorBizRepository.get(address); if (executorBiz != null) { return executorBiz; } // 创建获取代理对象（重点看getObject方法） executorBiz = (ExecutorBiz) new XxlRpcReferenceBean( NetEnum.NETTY_HTTP, Serializer.SerializeEnum.HESSIAN.getSerializer(), CallType.SYNC, LoadBalance.ROUND, ExecutorBiz.class, null, 5000, address, XxlJobAdminConfig.getAdminConfig().getAccessToken(), null, null).getObject(); //设置缓存 executorBizRepository.put(address, executorBiz); return executorBiz; } }
看看代理对象内部实现
public class XxlRpcReferenceBean { ...... //重点关注的方法, 被代理对象的run方法最终会到此对象的invoke public Object getObject() { return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[] { iface }, new InvocationHandler() { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { ...... // 组装RPC请求参数 XxlRpcRequest xxlRpcRequest = new XxlRpcRequest(); xxlRpcRequest.setRequestId(UUID.randomUUID().toString()); xxlRpcRequest.setCreateMillisTime(System.currentTimeMillis()); xxlRpcRequest.setAccessToken(accessToken); xxlRpcRequest.setClassName(className); xxlRpcRequest.setMethodName(methodName); xxlRpcRequest.setParameterTypes(parameterTypes); xxlRpcRequest.setParameters(parameters);
...... //最终都会通过此方法发起RPC //此处的client为上面创建代理对象时传入的NetEnum.NETTY_HTTP //即NettyHttpClient对象 //最终会通过netty来与执行器发起通信，细节不再继续追溯 client.asyncSend(finalAddress, xxlRpcRequest); ...... } }); } }
到此，xxl-job调度中心的初始化和调度执行流程，我们大概都知道了。那么，当调度中心向执行器发起调度请求时，执行器又是怎么做的呢？
那就还得再从执行器的初始化说起。
执行器
我们还是以spring boot版本的执行器为例。
初始化
首先会创建并初始化 XxlJobSpringExecutor实例，如下：
@Bean(initMethod = "start", destroyMethod = "destroy") public XxlJobSpringExecutor xxlJobExecutor() { logger.info(">>>>>>>>>>> xxl-job config init."); XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor(); xxlJobSpringExecutor.setAdminAddresses(adminAddresses); xxlJobSpringExecutor.setAppName(appName); xxlJobSpringExecutor.setIp(ip); xxlJobSpringExecutor.setPort(port); xxlJobSpringExecutor.setAccessToken(accessToken); xxlJobSpringExecutor.setLogPath(logPath); xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays); return xxlJobSpringExecutor; }
在初始化完成后会调用 XxlJobSpringExecutor 的start方法。
public class XxlJobSpringExecutor extends XxlJobExecutor implements ApplicationContextAware { @Override public void start() throws Exception { // JobHandler注解名与spring 托管的bean(我们的job)建立映射关系并缓存到Map initJobHandlerRepository(applicationContext); // 指定使用SpringGlueFactory， 不在我们本次探讨范围，暂时忽略 GlueFactory.refreshInstance(1);
// 调用父类XxlJobExecutor的start方法 super.start(); } }
我们看看initJobHandlerRepository方法。
private void initJobHandlerRepository(ApplicationContext applicationContext){ if (applicationContext == null) { return; }
// 获取带有@JobHandler修饰的bean Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(JobHandler.class); if (serviceBeanMap!=null && serviceBeanMap.size()>0) { for (Object serviceBean : serviceBeanMap.values()) { if (serviceBean instanceof IJobHandler){ //获取@JobHandler值 String name = serviceBean.getClass().getAnnotation(JobHandler.class).value(); IJobHandler handler = (IJobHandler) serviceBean; if (loadJobHandler(name) != null) { throw new RuntimeException("xxl-job jobhandler naming conflicts."); } //缓存到Map. 建立映射关系 registJobHandler(name, handler); } } } } public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){ return jobHandlerRepository.put(name, jobHandler); }
接下来看父类XxlJobExecutor 的start方法。
public class XxlJobExecutor { ...... public void start() throws Exception { // 设置job的日志目录 XxlJobFileAppender.initLogPath(logPath); // 初始化AdminBiz代理对象，该代理对象用于与调度中心进行RPC通信 initAdminBizList(adminAddresses, accessToken); // 日志清理线程 JobLogFileCleanThread.getInstance().start(logRetentionDays); // 回调线程（RPC回调到调度中心） TriggerCallbackThread.getInstance().start(); //启动服务并向调度中心发起注册请求 port = port>0?port: NetUtil.findAvailablePort(9999); ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp(); initRpcProvider(ip, port, appName, accessToken); } ...... }
其中，我们重点关注initAdminBizList 和 initRpcProvider 两个方法。
...... private static List<AdminBiz> adminBizList; private void initAdminBizList(String adminAddresses, String accessToken) throws Exception { serializer = Serializer.SerializeEnum.HESSIAN.getSerializer(); //如果是有多个调度中心地址，则创建多个实例 if (adminAddresses!=null && adminAddresses.trim().length()>0) { for (String address: adminAddresses.trim().split(",")) { if (address!=null && address.trim().length()>0) { //http://调度中心地址/api String addressUrl = address.concat(AdminBiz.MAPPING); //创建代理对象，似曾相识？ //这在我们讲调度中心的时候，已经讲过。 AdminBiz adminBiz = (AdminBiz) new XxlRpcReferenceBean( NetEnum.NETTY_HTTP, serializer, CallType.SYNC, LoadBalance.ROUND, AdminBiz.class, null, 10000, addressUrl, accessToken, null, null ).getObject(); if (adminBizList == null) { adminBizList = new ArrayList<AdminBiz>(); } //代理对象加入缓存 adminBizList.add(adminBiz); } } } } ......
接下来，我们再看看 initRpcProvider 这个最关键的方法之一，其包含了服务的启动。
...... private XxlRpcProviderFactory xxlRpcProviderFactory = null; private void initRpcProvider(String ip, int port, String appName, String accessToken) throws Exception { // 获取当前服务地址 (ip:port) String address = IpUtil.getIpPort(ip, port); //组装注册参数 Map<String, String> serviceRegistryParam = new HashMap<String, String>(); serviceRegistryParam.put("appName", appName); serviceRegistryParam.put("address", address); xxlRpcProviderFactory = new XxlRpcProviderFactory(); //最需要注意的是 //NetEnum.NETTY_HTTP指定使用NettyHttpServer作为我们的服务器 //ExecutorServiceRegistry为我们的服务注册的执行器 xxlRpcProviderFactory.initConfig(NetEnum.NETTY_HTTP, Serializer.SerializeEnum.HESSIAN.getSerializer(), ip, port, accessToken, ExecutorServiceRegistry.class, serviceRegistryParam); // add services xxlRpcProviderFactory.addService(ExecutorBiz.class.getName(), null, new ExecutorBizImpl()); // 启动服务，并向调度中心发起注册请求 xxlRpcProviderFactory.start(); } ......
接下来，我们直接看启动服务的方法。
public class XxlRpcProviderFactory { ...... private Server server; private ServiceRegistry serviceRegistry; private String serviceAddress; public void start() throws Exception { // 本（执行器）服务的地址 serviceAddress = IpUtil.getIpPort(this.ip, port); // 即上面指定的NettyHttpServer server = netType.serverClass.newInstance(); // 启动后回调此方法 server.setStartedCallback(new BaseCallback() { @Override public void run() throws Exception { if (serviceRegistryClass != null) { //即上面指定的ExecutorServiceRegistry serviceRegistry = serviceRegistryClass.newInstance(); // 向调度中心发起注册请求 serviceRegistry.start(serviceRegistryParam); if (serviceData.size() > 0) { serviceRegistry.registry(serviceData.keySet(), serviceAddress); } } } }); ...... //启动 server.start(this); } ...... }
以上，会启动NettyHttpServer服务， 通过设置启动回调来向调度中心发起注册请求。接下来，看看是怎么注册的。
@Override public void start(Map<String, String> param) { //调用ExecutorRegistryThread对象的start方法 ExecutorRegistryThread.getInstance() .start(param.get("appName"), param.get("address")); }
public class ExecutorRegistryThread { private static Logger logger = LoggerFactory.getLogger(ExecutorRegistryThread.class); private static ExecutorRegistryThread instance = new ExecutorRegistryThread(); public static ExecutorRegistryThread getInstance(){ return instance; } private Thread registryThread; private volatile boolean toStop = false; public void start(final String appName, final String address){ ...... registryThread = new Thread(new Runnable() { @Override public void run() { while (!toStop) { try { //注册参数 RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, address); for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) { try { //真正发起注册的方法 //adminBiz对象即为我们上面的代理对象 //触发的实际为代理对象的invoke方法 ReturnT<String> registryResult = adminBiz.registry(registryParam); if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) { registryResult = ReturnT.SUCCESS; logger.debug(">>>>>>>>>>> xxl-job registry success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult}); break; } else { logger.info(">>>>>>>>>>> xxl-job registry fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult}); } } catch (Exception e) { logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e); } } } catch (Exception e) { if (!toStop) { logger.error(e.getMessage(), e); } } try { if (!toStop) { //默认每隔30S触发一次注册 TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT); } } catch (InterruptedException e) { if (!toStop) { logger.warn(">>>>>>>>>>> xxl-job, executor registry thread interrupted, error msg:{}", e.getMessage()); } } } //移除注册信息 ........ }); registryThread.setDaemon(true); registryThread.setName("xxl-job, executor ExecutorRegistryThread"); //启动线程 registryThread.start(); } ...... public void toStop() { toStop = true; // interrupt and wait registryThread.interrupt(); try { registryThread.join(); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } }
以上，通过启动ExecutorRegistryThread线程进行注册，最终发起rpc请求的仍然是我们之前（调度中心）介绍的代理对象实例，就不作过多描述，该线程默认情况下会每隔30s发送心跳到调度中心。
以上即为主要初始化流程。那么，我们的执行中心到底是如何接收调度中心发起的调度请求的呢？
执行流程
在回到NettyHttpServer的启动流程。
public class NettyHttpServer extends Server {
private Thread thread;
@Override public void start(final XxlRpcProviderFactory xxlRpcProviderFactory) throws Exception { thread = new Thread(new Runnable() { @Override public void run() { ...... try { // start server ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new HttpServerCodec()); ch.pipeline().addLast(new HttpObjectAggregator(5*1024*1024)); //重点关注 ch.pipeline().addLast(new NettyHttpServerHandler(xxlRpcProviderFactory, serverHandlerPool)); } }).childOption(ChannelOption.SO_KEEPALIVE, true); ...... } ...... } }); thread.setDaemon(true); thread.start(); } ...... }
以上值得注意的是，在server启动时，会初始化NettyHttpServerHandler实例，当请求到来时，会到NettyHttpServerHandler的channelRead0方法。
public class NettyHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> { private static final Logger logger = LoggerFactory.getLogger(NettyHttpServerHandler.class); private XxlRpcProviderFactory xxlRpcProviderFactory; private ThreadPoolExecutor serverHandlerPool; public NettyHttpServerHandler(final XxlRpcProviderFactory xxlRpcProviderFactory, final ThreadPoolExecutor serverHandlerPool) { this.xxlRpcProviderFactory = xxlRpcProviderFactory; this.serverHandlerPool = serverHandlerPool; } //处理请求 @Override protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception { // request parse final byte[] requestBytes = ByteBufUtil.getBytes(msg.content()); final String uri = msg.uri(); final boolean keepAlive = HttpUtil.isKeepAlive(msg); // 通过线程池异步执行 serverHandlerPool.execute(new Runnable() { @Override public void run() { process(ctx, uri, requestBytes, keepAlive); } }); }
private void process(ChannelHandlerContext ctx, String uri, byte[] requestBytes, boolean keepAlive){ String requestId = null; try { if ("/services".equals(uri)) { // services mapping // request StringBuffer stringBuffer = new StringBuffer("<ui>"); for (String serviceKey: xxlRpcProviderFactory.getServiceData().keySet()) { stringBuffer.append("<li>").append(serviceKey).append(": ").append(xxlRpcProviderFactory.getServiceData().get(serviceKey)).append("</li>"); } stringBuffer.append("</ui>"); // response serialize byte[] responseBytes = stringBuffer.toString().getBytes("UTF-8"); // response-write writeResponse(ctx, keepAlive, responseBytes); } else { // valid if (requestBytes.length == 0) { throw new XxlRpcException("xxl-rpc request data empty."); } // request deserialize XxlRpcRequest xxlRpcRequest = (XxlRpcRequest) xxlRpcProviderFactory.getSerializer().deserialize(requestBytes, XxlRpcRequest.class); requestId = xxlRpcRequest.getRequestId(); // 处理请求 XxlRpcResponse xxlRpcResponse = xxlRpcProviderFactory.invokeService(xxlRpcRequest); // response serialize byte[] responseBytes = xxlRpcProviderFactory.getSerializer().serialize(xxlRpcResponse); // response-write writeResponse(ctx, keepAlive, responseBytes); } } catch (Exception e) { ...... } } ...... }
public XxlRpcResponse invokeService(XxlRpcRequest xxlRpcRequest) { ...... String serviceKey = makeServiceKey(xxlRpcRequest.getClassName(), xxlRpcRequest.getVersion()); //取出ExecutorBizImpl实例 Object serviceBean = serviceData.get(serviceKey); ...... try { // 反射调用ExecutorBizImpl对象run方法 Class<?> serviceClass = serviceBean.getClass(); String methodName = xxlRpcRequest.getMethodName(); Class<?>[] parameterTypes = xxlRpcRequest.getParameterTypes(); Object[] parameters = xxlRpcRequest.getParameters(); Method method = serviceClass.getMethod(methodName, parameterTypes); method.setAccessible(true); Object result = method.invoke(serviceBean, parameters); xxlRpcResponse.setResult(result); } catch (Throwable t) { // catch error logger.error("xxl-rpc provider invokeService error.", t); xxlRpcResponse.setErrorMsg(ThrowableUtil.toString(t)); } return xxlRpcResponse; }
public class ExecutorBizImpl implements ExecutorBiz { ...... @Override public ReturnT<String> run(TriggerParam triggerParam) { // 缓存获取JobThread对象 JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId()); IJobHandler jobHandler = jobThread!=null?jobThread.getHandler():null; String removeOldReason = null; GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType()); if (GlueTypeEnum.BEAN == glueTypeEnum) { // 缓存中获取IJobHandler对象（即我们的业务job） // 之前通过扫描注解存入缓存 IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler()); // valid old jobThread if (jobThread!=null && jobHandler != newJobHandler) { // change handler, need kill old thread removeOldReason = "change jobhandler or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { jobHandler = newJobHandler; if (jobHandler == null) { return new ReturnT<String>(ReturnT.FAIL_CODE, "job handler [" + triggerParam.getExecutorHandler() + "] not found."); } } } else if (GlueTypeEnum.GLUE_GROOVY == glueTypeEnum) { // valid old jobThread if (jobThread != null && !(jobThread.getHandler() instanceof GlueJobHandler && ((GlueJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) { // change handler or gluesource updated, need kill old thread removeOldReason = "change job source or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { try { //从DB中获取源码，通过groovy进行加载并实例化 IJobHandler originJobHandler = GlueFactory.getInstance().loadNewInstance(triggerParam.getGlueSource()); jobHandler = new GlueJobHandler(originJobHandler, triggerParam.getGlueUpdatetime()); } catch (Exception e) { logger.error(e.getMessage(), e); return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage()); } } } else if (glueTypeEnum!=null && glueTypeEnum.isScript()) { // valid old jobThread if (jobThread != null && !(jobThread.getHandler() instanceof ScriptJobHandler && ((ScriptJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) { // change script or gluesource updated, need kill old thread removeOldReason = "change job source or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { //读取脚本，写入文件，最终执行通过commons-exec jobHandler = new ScriptJobHandler(triggerParam.getJobId(), triggerParam.getGlueUpdatetime(), triggerParam.getGlueSource(), GlueTypeEnum.match(triggerParam.getGlueType())); } } else { return new ReturnT<String>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid."); } // 阻塞策略 if (jobThread != null) { ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null); if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) { // 丢弃后续调度 if (jobThread.isRunningOrHasQueue()) { return new ReturnT<String>(ReturnT.FAIL_CODE, "block strategy effect："+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle()); } } else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) { // 覆盖之前调度 if (jobThread.isRunningOrHasQueue()) { removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle(); jobThread = null; } } else { // just queue trigger } } // 第一次执行或者是覆盖之前调度策略 if (jobThread == null) { //开启线程，执行任务 jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason); } // 触发任务入队 ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam); return pushResult; } }
至此，我们离真相只差最后一步，最后再看看XxlJobExecutor.registJobThread
...... private static ConcurrentHashMap<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<Integer, JobThread>(); public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){ //新线程执行 JobThread newJobThread = new JobThread(jobId, handler); //线程执行 newJobThread.start(); logger.info(">>>>>>>>>>> xxl-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler}); //放入缓存 JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread); if (oldJobThread != null) { //旧任务线程停止，覆盖策略 oldJobThread.toStop(removeOldReason); oldJobThread.interrupt(); } return newJobThread; } ......
public class JobThread extends Thread{ private static Logger logger = LoggerFactory.getLogger(JobThread.class); private int jobId; private IJobHandler handler; private LinkedBlockingQueue<TriggerParam> triggerQueue; private Set<Integer> triggerLogIdSet; // avoid repeat trigger for the same TRIGGER_LOG_ID private volatile boolean toStop = false; private String stopReason; private boolean running = false; // if running job private int idleTimes = 0; // idel times
public JobThread(int jobId, IJobHandler handler) { this.jobId = jobId; this.handler = handler; this.triggerQueue = new LinkedBlockingQueue<TriggerParam>(); this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<Integer>()); } public IJobHandler getHandler() { return handler; } /** * trigger入队，执行的时候出队 * * @param triggerParam * @return */ public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) { // avoid repeat if (triggerLogIdSet.contains(triggerParam.getLogId())) { logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId()); return new ReturnT<String>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId()); } triggerLogIdSet.add(triggerParam.getLogId()); triggerQueue.add(triggerParam); return ReturnT.SUCCESS; } /** * kill job thread * * @param stopReason */ public void toStop(String stopReason) { /** * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)， * 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身； * 所以需要注意，此处彻底销毁本线程，需要通过共享变量方式； */ this.toStop = true; this.stopReason = stopReason; } /** * is running job * @return */ public boolean isRunningOrHasQueue() { return running || triggerQueue.size()>0; } @Override public void run() { // init try { handler.init(); } catch (Throwable e) { logger.error(e.getMessage(), e); } // execute while(!toStop){ running = false; idleTimes++; TriggerParam triggerParam = null; ReturnT<String> executeResult = null; try { //出队消费，3秒获取不到就返回null triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS); if (triggerParam!=null) { running = true; idleTimes = 0; triggerLogIdSet.remove(triggerParam.getLogId()); // 日志 "logPath/yyyy-MM-dd/9999.log" String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTim()), triggerParam.getLogId()); XxlJobFileAppender.contextHolder.set(logFileName); //任务分片数据 ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(triggerParam.getBroadcastIndex(), triggerParam.getBroadcastTotal())); // execute XxlJobLogger.log("<br>----------- xxl-job job execute start -----------<br>----------- Param:" + triggerParam.getExecutorParams()); if (triggerParam.getExecutorTimeout() > 0) { //有超时限制 Thread futureThread = null; try { final TriggerParam triggerParamTmp = triggerParam; FutureTask<ReturnT<String>> futureTask = new FutureTask<ReturnT<String>>(new Callable<ReturnT<String>>() { @Override public ReturnT<String> call() throws Exception { //执行业务job return handler.execute(triggerParamTmp.getExecutorParams()); } }); futureThread = new Thread(futureTask); futureThread.start(); //可能超时 executeResult = futureTask.get(triggerParam.getExecutorTimeout(), TimeUnit.SECONDS); } catch (TimeoutException e) { XxlJobLogger.log("<br>----------- xxl-job job execute timeout"); XxlJobLogger.log(e); executeResult = new ReturnT<String>(IJobHandler.FAIL_TIMEOUT.getCode(), "job execute timeout "); } finally { futureThread.interrupt(); } } else { // 无超时限制的，直接执行 executeResult = handler.execute(triggerParam.getExecutorParams()); } ......
// destroy try { handler.destroy(); } catch (Throwable e) { logger.error(e.getMessage(), e); } logger.info(">>>>>>>>>>> xxl-job JobThread stoped, hashCode:{}", Thread.currentThread()); } }
我们的业务基本，都是实现IJobHandler的excute方法，因此，最终就会到我们的业务方法。
到此，我们的xxl-job之旅就暂且告一段落。其实其中还有不少内容值得去深探，有兴趣的可以继续去看看。