透过源码看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之旅就暂且告一段落。其实其中还有不少内容值得去深探，有兴趣的可以继续去看看。