首页 > 代码库 > Quartz任务调度源码分析
Quartz任务调度源码分析
从源码分析中可以看出,任务的整个调度过程为,初始化线程池,及调度器QuartzScheduler,然后由线程池去执行QuartzSchedulerThread,将触发器任务(job与触发器)添加到存储器(TreeSet,timeTrriger)中,然后启动调度器,QuartzSchedulerThread从timeTrriger去除待触发的任务,并包装成TriggerFiredBundle,然后由JobRunShellFactory
创建TriggerFiredBundle的执行线程JobRunShell, 调度执行通过线程池SimpleThreadPool去执行JobRunShell,而JobRunShell执行的就是job.execute(JobExecutionContext context)。Quartz主要中的集合类有ArrayList,LinkedList,HashMap,TreeSet(TreeMap);之所以用到上面四个集合类,主要用到集合的如下特点:ArrayList访问速度快,LinkedList添加删除元素快;HashMap添加删除快,TreeSet访问速度快。
触发任务创建工厂类
Java代码 下载
public class JTAJobRunShellFactory
implements JobRunShellFactory
{
public void initialize(Scheduler sched)
throws SchedulerConfigException
{
scheduler = sched;
}
public JobRunShell createJobRunShell(TriggerFiredBundle bundle)
throws SchedulerException
{
return new JTAJobRunShell(scheduler, bundle);
}
private Scheduler scheduler;
}
//触发任务运行类
Java代码 下载
public class JTAJobRunShell extends JobRunShell
{
public JTAJobRunShell(Scheduler scheduler, TriggerFiredBundle bndle)
{
super(scheduler, bndle);
transactionTimeout = null;
}
}
public class JobRunShell extends SchedulerListenerSupport
implements Runnable
{
public JobRunShell(Scheduler scheduler, TriggerFiredBundle bndle)
{
jec = null;
qs = null;
firedTriggerBundle = null;
this.scheduler = null;
shutdownRequested = false;
this.scheduler = scheduler;
firedTriggerBundle = bndle;
}
public void run()
{
//添加到内部监听器
qs.addInternalSchedulerListener(this);
label0:
{
//protected JobExecutionContextImpl jec,job执行上下文
OperableTrigger trigger = (OperableTrigger)jec.getTrigger();
JobDetail jobDetail = jec.getJobDetail();
org.quartz.Trigger.CompletedExecutionInstruction instCode;
do
{
JobExecutionException jobExEx = null;
Job job = jec.getJobInstance();
try
{
begin();
}
catch(SchedulerException se)
{
qs.notifySchedulerListenersError((new StringBuilder()).append("Error executing Job (").append(jec.getJobDetail().getKey()).append(": couldn‘t begin execution.").toString(), se);
break label0;
}
try
{
if(!notifyListenersBeginning(jec))
break label0;
}
catch(VetoedException ve)
{
try
{
org.quartz.Trigger.CompletedExecutionInstruction instCode = trigger.executionComplete(jec, null);
qs.notifyJobStoreJobVetoed(trigger, jobDetail, instCode);
if(jec.getTrigger().getNextFireTime() == null)
qs.notifySchedulerListenersFinalized(jec.getTrigger());
complete(true);
}
catch(SchedulerException se)
{
qs.notifySchedulerListenersError((new StringBuilder()).append("Error during veto of Job (").append(jec.getJobDetail().getKey()).append(": couldn‘t finalize execution.").toString(), se);
}
break label0;
}
long startTime = System.currentTimeMillis();
long endTime = startTime;
try
{
log.debug((new StringBuilder()).append("Calling execute on job ").append(jobDetail.getKey()).toString());
//执行Job,关键
job.execute(jec);
endTime = System.currentTimeMillis();
}
catch(JobExecutionException jee)
{
endTime = System.currentTimeMillis();
jobExEx = jee;
getLog().info((new StringBuilder()).append("Job ").append(jobDetail.getKey()).append(" threw a JobExecutionException: ").toString(), jobExEx);
}
catch(Throwable e)
{
endTime = System.currentTimeMillis();
getLog().error((new StringBuilder()).append("Job ").append(jobDetail.getKey()).append(" threw an unhandled Exception: ").toString(), e);
SchedulerException se = new SchedulerException("Job threw an unhandled exception.", e);
qs.notifySchedulerListenersError((new StringBuilder()).append("Job (").append(jec.getJobDetail().getKey()).append(" threw an exception.").toString(), se);
jobExEx = new JobExecutionException(se, false);
}
//设置jJobExecutionContext运行时间
jec.setJobRunTime(endTime - startTime);
if(!notifyJobListenersComplete(jec, jobExEx))
break label0;
instCode = org.quartz.Trigger.CompletedExecutionInstruction.NOOP;
try
{
instCode = trigger.executionComplete(jec, jobExEx);
}
catch(Exception e)
{
SchedulerException se = new SchedulerException("Trigger threw an unhandled exception.", e);
qs.notifySchedulerListenersError("Please report this error to the Quartz developers.", se);
}
if(!notifyTriggerListenersComplete(jec, instCode))
break label0;
if(instCode == org.quartz.Trigger.CompletedExecutionInstruction.RE_EXECUTE_JOB)
{
jec.incrementRefireCount();
try
{
complete(false);
}
catch(SchedulerException se)
{
qs.notifySchedulerListenersError((new StringBuilder()).append("Error executing Job (").append(jec.getJobDetail().getKey()).append(": couldn‘t finalize execution.").toString(), se);
}
continue;
}
try
{
complete(true);
break;
}
catch(SchedulerException se)
{
qs.notifySchedulerListenersError((new StringBuilder()).append("Error executing Job (").append(jec.getJobDetail().getKey()).append(": couldn‘t finalize execution.").toString(), se);
}
} while(true);
//通知job执行完成
qs.notifyJobStoreJobComplete(trigger, jobDetail, instCode);
}
qs.removeInternalSchedulerListener(this);
break MISSING_BLOCK_LABEL_710;
Exception exception;
exception;
qs.removeInternalSchedulerListener(this);
throw exception;
}
protected JobExecutionContextImpl jec;//job执行上下文
protected QuartzScheduler qs;
protected TriggerFiredBundle firedTriggerBundle;
protected Scheduler scheduler;
protected volatile boolean shutdownRequested;
private final Logger log = LoggerFactory.getLogger(getClass());
}
//TriggerKey,JobKey包装类
Java代码 下载
class TriggerWrapper
{
TriggerWrapper(OperableTrigger trigger)
{
state = 0;
if(trigger == null)
{
throw new IllegalArgumentException("Trigger cannot be null!");
} else
{
this.trigger = trigger;
key = trigger.getKey();
jobKey = trigger.getJobKey();
return;
}
}
public boolean equals(Object obj)
{
if(obj instanceof TriggerWrapper)
{
TriggerWrapper tw = (TriggerWrapper)obj;
if(tw.key.equals(key))
return true;
}
return false;
}
public int hashCode()
{
return key.hashCode();
}
public OperableTrigger getTrigger()
{
return trigger;
}
public final TriggerKey key;
public final JobKey jobKey;
public final OperableTrigger trigger;
public int state;
public static final int STATE_WAITING = 0;//等待
public static final int STATE_ACQUIRED = 1;//就绪
public static final int STATE_EXECUTING = 2;//执行
public static final int STATE_COMPLETE = 3;//完成
public static final int STATE_PAUSED = 4;//暂停
public static final int STATE_BLOCKED = 5;//阻塞
public static final int STATE_PAUSED_BLOCKED = 6;//暂停阻塞
public static final int STATE_ERROR = 7;//错误
}
//简单触发器
Java代码 下载
public class SimpleTriggerImpl extends AbstractTrigger
implements SimpleTrigger, CoreTrigger
{
//获取下一次触发时间
public Date getNextFireTime()
{
return nextFireTime;
}
private Date startTime;
private Date endTime;
private Date nextFireTime;
private Date previousFireTime;
private int repeatCount;
private long repeatInterval;
private int timesTriggered;
private boolean complete;
}
//触发任务包装类
Java代码 下载
public class TriggerFiredBundle
implements Serializable
{
public TriggerFiredBundle(JobDetail job, OperableTrigger trigger, Calendar cal, boolean jobIsRecovering, Date fireTime, Date scheduledFireTime, Date prevFireTime,
Date nextFireTime)
{
this.job = job;
this.trigger = trigger;
this.cal = cal;
this.jobIsRecovering = jobIsRecovering;
this.fireTime = fireTime;
this.scheduledFireTime = scheduledFireTime;
this.prevFireTime = prevFireTime;
this.nextFireTime = nextFireTime;
}
private JobDetail job;
private OperableTrigger trigger;
private Calendar cal;
private boolean jobIsRecovering;
private Date fireTime;
private Date scheduledFireTime;
private Date prevFireTime;
private Date nextFireTime;
}
//触发任务包装结果类
Java代码 下载
public class TriggerFiredResult
{
public TriggerFiredResult(TriggerFiredBundle triggerFiredBundle)
{
this.triggerFiredBundle = triggerFiredBundle;
}
private TriggerFiredBundle triggerFiredBundle;
private Exception exception;
}
Quartz任务调度源码分析