quartz 2.2.x 源码学习 基本执行流程分析

quartz 官网中给出了一些基本概念，请先阅读官网相关概念。

http://www.quartz-scheduler.org/documentation/quartz-2.2.x/tutorials/

下面对最简单的一个任务调度工作进行分析，下面的代码每隔三秒中不断重复执行任务SimpleJob。

public class JobStart {

public static void main(String[] args) throws SchedulerException {

SchedulerFactory sf = new StdSchedulerFactory();

Scheduler sched = sf.getScheduler();

sched.clear();

String instanceId = sched.getSchedulerInstanceId();

JobDetail detail = JobBuilder.newJob(SimpleJob.class)

.withIdentity("Ins Id " + instanceId, instanceId)

.requestRecovery()

.build();

Trigger trigger = TriggerBuilder.newTrigger()

.withIdentity("Ins Id " + instanceId,instanceId)

.withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(3).repeatForever()).build();

sched.scheduleJob(detail, trigger);

sched.start();

}

}

@DisallowConcurrentExecution

public class SimpleJob implements Job{

private static Logger _log = LoggerFactory.getLogger(SimpleJob.class);

@Override

public void execute(JobExecutionContext context) throws JobExecutionException {

if(context.isRecovering()){

_log.info("is recovering");

}else{

_log.info("not recovering");

}

try {

_log.info("do time consumed job start");

Thread.sleep(5000);

_log.info("do time consumed job end");

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

quartz 执行任务可以总结为创建JobDetail描述待执行任务,创建Trigger描述什么时候去触发任务。最后将JobDetail与Trigger注册在调度器中，调度器启动后遍开始对任务调度。

因此了解其基本过程需要明白 创建调度器完成哪些工作？调度工作什么时候开始？任务如何被执行？

示例代码中，首先获取一个调度器。

Scheduler sched = sf.getScheduler();

1

进入该方法内部

org.quartz.impl.StdSchedulerFactory.getScheduler()

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;

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

获取调度器对象时，首先查找缓存，如果有就直接拿，若不存在则调用instantiate()方法来创建调度器对象。

进入instantiate方法内部，定义了许多变量。

org.quartz.impl.StdSchedulerFactory.instantiate()

private Scheduler instantiate() throws SchedulerException {

if (cfg == null) {

initialize();

}

if (initException != null) {

throw initException;

}

JobStore js = null;

ThreadPool tp = null;

QuartzScheduler qs = null;

DBConnectionManager dbMgr = null;

String instanceIdGeneratorClass = null;

Properties tProps = null;

......

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

ThreadPool tp为线程池对象，在该方法内部可以看到该线程池的创建 是通过从属性文件quartz.properties中拿到org.quartz.threadPool所定义的线城池类来创建。本文使用的是 org.quartz.simpl.SimpleThreadPool。而最终任务通过线程池来管理。

public static final String PROP_THREAD_POOL_PREFIX = "org.quartz.threadPool";

String tpClass = cfg.getStringProperty(PROP_THREAD_POOL_CLASS, SimpleThreadPool.class.getName());

if (tpClass == null) {

initException = new SchedulerException(

"ThreadPool class not specified. ");

throw initException;

}

try {

tp = (ThreadPool) loadHelper.loadClass(tpClass).newInstance();

} catch (Exception e) {

initException = new SchedulerException("ThreadPool class '"

+ tpClass + "' could not be instantiated.", e);

throw initException;

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

后续代码中会创建QuartzScheduler对象，最终来启动调动线程。

qs = new QuartzScheduler(rsrcs, idleWaitTime, dbFailureRetry);

1

在该构造器内部，会创建QuartzSchedulerThread对象，而QuartzSchedulerThread负责对我们定义的Job进行调度。当创建完后

提交给执行器对象schedThreadExecutor.execute(this.schedThread);准备执行。

public QuartzScheduler(QuartzSchedulerResources resources, long idleWaitTime, @Deprecated long dbRetryInterval)

throws SchedulerException {

this.resources = resources;

if (resources.getJobStore() instanceof JobListener) {

addInternalJobListener((JobListener)resources.getJobStore());

}

this.schedThread = new QuartzSchedulerThread(this, resources);

ThreadExecutor schedThreadExecutor = resources.getThreadExecutor();

schedThreadExecutor.execute(this.schedThread);

if (idleWaitTime > 0) {

this.schedThread.setIdleWaitTime(idleWaitTime);

}

1

2

3

4

5

6

7

8

9

10

11

12

13

当sf.getScheduler();执行完毕后，线程池，执行器，调度器，负责调度Job的线程都已创建完毕。而任务调度线程QuartzSchedulerThread也准备执行。

可以看到sf.getScheduler();调用完毕后，任务调度线程处于暂停状态，并不断检查状态等待恢复并执行任务。

org.quartz.core.QuartzSchedulerThread.run()

while (!halted.get()) {

try {

// check if we're supposed to pause...

synchronized (sigLock) {

while (paused && !halted.get()) {

try {

// wait until togglePause(false) is called...

sigLock.wait(1000L);

} catch (InterruptedException ignore) {

}

}

if (halted.get()) {

break;

}

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

在 QuartzSchedulerThread 构造方法中会设置两个变量初始值。

paused = true;

halted = new AtomicBoolean(false);

1

2

当我们获取到调度器对象，并指定trigger与detail后，调用start方法。

当调用执行器的 org.quartz.core.QuartzScheduler.start()方法在该方法内部会调用方法 schedThread.togglePause(false);将paused变量设置为false,并唤醒QuartzSchedulerThread线程。随QuartzSchedulerThread线程开始执行调度任务。

org.quartz.core.QuartzScheduler.start()

void togglePause(boolean pause) {

synchronized (sigLock) {

paused = pause;

if (paused) {

signalSchedulingChange(0);

} else {

sigLock.notifyAll();

}

}

1

2

3

4

5

6

7

8

9

10

11

12

至此任务调度工作准备开始，Job等待被调用。

QuartzSchedulerThread的run方法中检测到变量状态被修改，开始调度工作。

public void run() {

boolean lastAcquireFailed = false;

while (!halted.get()) {

try {

// check if we're supposed to pause...

synchronized (sigLock) {

while (paused && !halted.get()) {

try {

// wait until togglePause(false) is called...

sigLock.wait(1000L);

} catch (InterruptedException ignore) {

}

}

if (halted.get()) {

break;

}

}

int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();

if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads...

List<OperableTrigger> triggers = null;

long now = System.currentTimeMillis();

clearSignaledSchedulingChange();

... 省略

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

run方法中的执行过程可粗略归纳为

1.查找将要触发的Trigger

2.检测条件，等待执行

3.执行任务。

4.释放Trigger

在执行时，任务被包装为JobRunShell来运行。

当qsRsrcs.getThreadPool().runInThread(shell)被调用时，我们自己的Job被提交，并等待线程池调度执行。

JobRunShell shell = null;

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) {

// this case should never happen, as it is indicative of the

// scheduler being shutdown or a bug in the thread pool or

// a thread pool being used concurrently - which the docs

// say not to do...

getLog().error("ThreadPool.runInThread() return false!");

qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR);

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

在 qsRsrcs.getThreadPool().runInThread(shell)方法中会创建一个线程来运行我们定义的Job，而Job的信息则是通过JobDetail来获取。经过上述过程就形成了quartz 进行任务调度的基本过程。

最后总结，启动时的三个主要方法，及工作概述如下。

Scheduler sched = sf.getScheduler();

完成了线程池创建，调度线程创建，调度线程初始处理暂停状态。

sched.scheduleJob(detail, trigger);

指定JobDetail与Trigger。

sched.start();

修改状态未，使调度线程开始运行。任务以JobRunShell的形式被执行。

工艺流程图图