quartz群调查调度机制和源代码分析

public class HelloWorldMain
{

    Log
log = LogFactory.getLog(HelloWorldMain.class);

    public void run()
{

        try {

            //取得Schedule对象

            SchedulerFactory
sf = new StdSchedulerFactory();

            Scheduler
sch = sf.getScheduler(); 

            JobDetail
jd = new JobDetail("HelloWorldJobDetail",Scheduler.DEFAULT_GROUP,HelloWorldJob.class);

            Trigger
tg = TriggerUtils.makeMinutelyTrigger(1);

            tg.setName("HelloWorldTrigger");

            sch.scheduleJob(jd,
tg);

            sch.start();

        } catch (
Exception e ) {

            e.printStackTrace();

        }

    }

    public static void main(String[]
args) {

        HelloWorldMain
hw = new HelloWorldMain();

        hw.run();

    }

}

SchedulerFactory
sf = new StdSchedulerFactory();

Scheduler
sch = sf.getScheduler(); 

sch.start();

public Scheduler
getScheduler() throws SchedulerException
{

        if (cfg
== null)
{

            initialize();

        }

        SchedulerRepository
schedRep = SchedulerRepository.getInstance();

        //从"调度器仓库"中依据properties的SchedulerName配置获取一个调度器实例

        Scheduler
sched = schedRep.lookup(getSchedulerName());

        if (sched
!= null)
{

            if (sched.isShutdown())
{

                schedRep.remove(getSchedulerName());

            } else {

                return sched;

            }

        }

        //初始化调度器

        sched
= instantiate();

        return sched;

    }

public void initialize(ClassLoadHelper
loadHelper,

            SchedulerSignaler
signaler) throws SchedulerConfigException
{

        if (dsName
== null)
{

            throw new SchedulerConfigException("DataSource
name not set.");

        }

        classLoadHelper
= loadHelper;

        if(isThreadsInheritInitializersClassLoadContext())
{

            log.info("JDBCJobStore
threads will inherit ContextClassLoader of thread: " +
Thread.currentThread().getName());

            initializersLoader
= Thread.currentThread().getContextClassLoader();

        }

        this.schedSignaler
= signaler;

        //
If the user hasn't specified an explicit lock handler, then

        //
choose one based on CMT/Clustered/UseDBLocks.

        if (getLockHandler()
== null)
{

            //
If the user hasn't specified an explicit lock handler,

            //
then we *must* use DB locks with clustering

            if (isClustered())
{

                setUseDBLocks(true);

            }

            if (getUseDBLocks())
{

                if(getDriverDelegateClass()
!= null &&
getDriverDelegateClass().equals(MSSQLDelegate.class.getName()))
{

                    if(getSelectWithLockSQL()
== null)
{

                        //读取数据库LOCKS表中相应当前调度器的锁信息

                        String
msSqlDflt = "SELECT
* FROM {0}LOCKS WITH (UPDLOCK,ROWLOCK) WHERE " +
COL_SCHEDULER_NAME + "
= {1} AND LOCK_NAME = ?";

                        getLog().info("Detected
usage of MSSQLDelegate class - defaulting 'selectWithLockSQL' to '" +
msSqlDflt + "'.");

                        setSelectWithLockSQL(msSqlDflt);

                    }

                }

                getLog().info("Using
db table-based data access locking (synchronization).");

                setLockHandler(new StdRowLockSemaphore(getTablePrefix(),
getInstanceName(), getSelectWithLockSQL()));

            } else {

                getLog().info(

                    "Using
thread monitor-based data access locking (synchronization).");

                setLockHandler(new SimpleSemaphore());

            }

        }

    }

public void start() throws SchedulerException
{

        if (shuttingDown||
closed) {

            throw new SchedulerException(

                    "The
Scheduler cannot be restarted after shutdown() has been called.");

        }

        //
QTZ-212 : calling new schedulerStarting() method on the listeners

        //
right after entering start()

        //通知该调度器的listener启动開始

        notifySchedulerListenersStarting();

        if (initialStart
== null)
{

            initialStart
= new Date();

            //启动调度器的线程

            this.resources.getJobStore().schedulerStarted();          
 

            //启动plugins

            startPlugins();

        } else {

            resources.getJobStore().schedulerResumed();

        }

        schedThread.togglePause(false);

        getLog().info(

                "Scheduler
" +
resources.getUniqueIdentifier() + "
started.");

        //通知该调度器的listener启动完毕

        notifySchedulerListenersStarted();

    }

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();

                    try {

                        //调度器在trigger队列中寻找30秒内一定数目的trigger准备运行调度,

                        //參数1:nolaterthan
= now+3000ms,參数2 最大获取数量,大小取线程池线程剩余量与定义值得较小者

                        //參数3
时间窗体 默觉得0,程序会在nolaterthan后加上窗体大小来选择trigger

                        triggers
= qsRsrcs.getJobStore().acquireNextTriggers(

                                now
+ idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());

                        //上一步获取成功将失败标志置为false;

                        lastAcquireFailed
= false;

                        if (log.isDebugEnabled())

                            log.debug("batch
acquisition of " +
(triggers == null ? 0 :
triggers.size()) + "
triggers");

                    } catch (JobPersistenceException
jpe) {

                        if(!lastAcquireFailed)
{

                            qs.notifySchedulerListenersError(

                                "An
error occurred while scanning for the next triggers to fire.",

                                jpe);

                        }

                        //捕捉到异常则值标志为true,再次尝试获取

                        lastAcquireFailed
= true;

                        continue;

                    } catch (RuntimeException
e) {

                        if(!lastAcquireFailed)
{

                            getLog().error("quartzSchedulerThreadLoop:
RuntimeException "

                                    +e.getMessage(),
e);

                        }

                        lastAcquireFailed
= true;

                        continue;

                    }

                    if (triggers
!= null &&
!triggers.isEmpty()) {

                        now
= System.currentTimeMillis();

                        long triggerTime
= triggers.get(0).getNextFireTime().getTime();

                        long timeUntilTrigger
= triggerTime - now;//计算距离trigger触发的时间

                        while(timeUntilTrigger
> 2)
{

                            synchronized (sigLock)
{

                                if (halted.get())
{

                                    break;

                                }

                                //假设这时调度器发生了改变,新的trigger加入进来,那么有可能新加入的trigger比当前待运行的trigger

                                //更急迫,那么须要放弃当前trigger又一次获取,然而,这里存在一个值不值得的问题,假设又一次获取新trigger

                                //的时间要长于当前时间到新trigger出发的时间,那么即使放弃当前的trigger,仍然会导致xntrigger获取失败,

                                //但我们又不知道获取新的trigger须要多长时间,于是,我们做了一个主观的评判,若jobstore为RAM,那么

                                //假定获取时间须要7ms,若jobstore是持久化的,假定其须要70ms,当前时间与新trigger的触发时间之差小于

                                //
这个值的我们觉得不值得又一次获取,返回false

                                //这里推断是否有上述情况发生,值不值得放弃本次trigger,若判定不放弃,则线程直接等待至trigger触发的时刻

                                if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false))
{

                                    try {

                                        //
we could have blocked a long while

                                        //
on 'synchronize', so we must recompute

                                        now
= System.currentTimeMillis();

                                        timeUntilTrigger
= triggerTime - now;

                                        if(timeUntilTrigger
>= 1)

                                            sigLock.wait(timeUntilTrigger);

                                    } catch (InterruptedException
ignore) {

                                    }

                                }

                            }

                            //该方法调用了上面的判定方法,作为再次判定的逻辑

                            //到达这里有两种情况1.决定放弃当前trigger,那么再判定一次,假设仍然有放弃,那么清空triggers列表并

                            //
退出循环 2.不放弃当前trigger,且线程已经wait到trigger触发的时刻,那么什么也不做

                            if(releaseIfScheduleChangedSignificantly(triggers,
triggerTime)) {

                                break;

                            }

                            now
= System.currentTimeMillis();

                            timeUntilTrigger
= triggerTime - now;

                            //这时触发器已经即将触发,值会<2

                        }

                        //
this happens if releaseIfScheduleChangedSignificantly decided to release triggers

                        if(triggers.isEmpty())

                            continue;

                        //
set triggers to 'executing'

                        List<TriggerFiredResult>
bndles = new ArrayList<TriggerFiredResult>();

                        boolean goAhead
= true;

                        synchronized(sigLock)
{

                            goAhead
= !halted.get();

                        }

                        if(goAhead)
{

                            try {

                                //触发triggers,结果付给bndles,注意,从这里返回后,trigger在数据库中已经经过了锁定,解除锁定,这一套过程

                                //所以说,quratz定不是等到job运行完才释放trigger资源的占有,而是读取完本次触发所需的信息后马上释放资源

                                //然后再运行jobs

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

                                //QTZ-179
: a problem occurred interacting with the triggers from the db

                                //we
release them and loop again

                                for (int i
= 0;
i < triggers.size(); i++) {

                                    qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));

                                }

                                continue;

                            }

                        }

                        //迭代trigger的信息,分别跑job

                        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;

                            }

                            //
it's possible to get 'null' if the triggers was paused,

                            //
blocked, or other similar occurrences that prevent it being

                            //
fired at this time...  or if the scheduler was shutdown (halted)

                            //在特殊情况下,bndle可能为null,看triggerFired方法能够看到,当从数据库获取trigger时,假设status不是

                            //STATE_ACQUIRED,那么会直接返回空.quratz这样的情况下本调度器启动重试流程,又一次获取4次,若仍有问题,

                            //
则抛出异常.

                            if (bndle
== null)
{

                                qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));

                                continue;

                            }

                            //运行job

                            JobRunShell
shell = null;

                            try {

                                //创建一个job的Runshell

                                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;

                            }

                            //把runShell放在线程池里跑

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

                            }

                        }

                        continue; //
while (!halted)

                    }

                } else { //
if(availThreadCount > 0)

                    //
should never happen, if threadPool.blockForAvailableThreads() follows contract

                    continue; //
while (!halted)

                }

                //保证负载平衡的方法,每次运行一轮触发后,本scheduler会等待一个随机的时间,这样就使得其它节点上的scheduler能够得到资源.

                long now
= System.currentTimeMillis();

                long waitTime
= now + getRandomizedIdleWaitTime();

                long timeUntilContinue
= waitTime - now;

                synchronized(sigLock)
{

                    try {

                      if(!halted.get())
{

                        //
QTZ-336 A job might have been completed in the mean time and we might have

                        //
missed the scheduled changed signal by not waiting for the notify() yet

                        //
Check that before waiting for too long in case this very job needs to be

                        //
scheduled very soon

                        if (!isScheduleChanged())
{

                          sigLock.wait(timeUntilContinue);

                        }

                      }

                    } catch (InterruptedException
ignore) {

                    }

                }

            } catch(RuntimeException
re) {

                getLog().error("Runtime
error occurred in main trigger firing loop.",
re);

            }

        } //
while (!halted)

        //
drop references to scheduler stuff to aid garbage collection...

        qs
= null;

        qsRsrcs
= null;

    }

triggers
= qsRsrcs.getJobStore().acquireNextTriggers(

now
+ idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());

public List<OperableTrigger>
acquireNextTriggers(final long noLaterThan, final int maxCount, final long timeWindow)

        throws JobPersistenceException
{

        String
lockName;

        if(isAcquireTriggersWithinLock()
|| maxCount > 1)
{

            lockName
= LOCK_TRIGGER_ACCESS;

        } else {

            lockName
= null;

        }

        return executeInNonManagedTXLock(lockName,

                new TransactionCallback<List<OperableTrigger>>()
{

                    public List<OperableTrigger>
execute(Connection conn) throws JobPersistenceException
{

                        return acquireNextTrigger(conn,
noLaterThan, maxCount, timeWindow);

                    }

                },

                new TransactionValidator<List<OperableTrigger>>()
{

                    public Boolean
validate(Connection conn, List<OperableTrigger> result) throws JobPersistenceException
{

                        //...异常处理回调方法

                    }

                });

    }

public static final String
SELECT_FOR_LOCK = "SELECT
* FROM "

            +
TABLE_PREFIX_SUBST + TABLE_LOCKS + "
WHERE " +
COL_SCHEDULER_NAME + "
= " +
SCHED_NAME_SUBST

            + "
AND " +
COL_LOCK_NAME + "
= ? FOR UPDATE";

public List<TriggerFiredResult>
triggersFired(final List<OperableTrigger>
triggers) throws JobPersistenceException
{

        return executeInNonManagedTXLock(LOCK_TRIGGER_ACCESS,

                new TransactionCallback<List<TriggerFiredResult>>()
{

                    public List<TriggerFiredResult>
execute(Connection conn) throws JobPersistenceException
{

                        List<TriggerFiredResult>
results = new ArrayList<TriggerFiredResult>();

                        TriggerFiredResult
result;

                        for (OperableTrigger
trigger : triggers) {

                            try {

                              TriggerFiredBundle
bundle = triggerFired(conn, trigger);

                              result
= new TriggerFiredResult(bundle);

                            } catch (JobPersistenceException
jpe) {

                                result
= new TriggerFiredResult(jpe);

                            } catch(RuntimeException
re) {

                                result
= new TriggerFiredResult(re);

                            }

                            results.add(result);

                        }

                        return results;

                    }

                },

                new TransactionValidator<List<TriggerFiredResult>>()
{

                    @Override

                    public Boolean
validate(Connection conn, List<TriggerFiredResult> result) throws JobPersistenceException
{

                        //...异常处理回调方法

                    }

                });

    }

protected TriggerFiredBundle
triggerFired(Connection conn,

            OperableTrigger
trigger)

        throws JobPersistenceException
{

        JobDetail
job;

        Calendar
cal = null;

        //
Make sure trigger wasn't deleted, paused, or completed...

        try { //
if trigger was deleted, state will be STATE_DELETED

            String
state = getDelegate().selectTriggerState(conn,

                    trigger.getKey());

            if (!state.equals(STATE_ACQUIRED))
{

                return null;

            }

        } catch (SQLException
e) {

            throw new JobPersistenceException("Couldn't
select trigger state: "

                    +
e.getMessage(), e);

        }

        try {

            job
= retrieveJob(conn, trigger.getJobKey());

            if (job
== null)
{ return null;
}

        } catch (JobPersistenceException
jpe) {

            try {

                getLog().error("Error
retrieving job, setting trigger state to ERROR.",
jpe);

                getDelegate().updateTriggerState(conn,
trigger.getKey(),

                        STATE_ERROR);

            } catch (SQLException
sqle) {

                getLog().error("Unable
to set trigger state to ERROR.",
sqle);

            }

            throw jpe;

        }

        if (trigger.getCalendarName()
!= null)
{

            cal
= retrieveCalendar(conn, trigger.getCalendarName());

            if (cal
== null)
{ return null;
}

        }

        try {

            getDelegate().updateFiredTrigger(conn,
trigger, STATE_EXECUTING, job);

        } catch (SQLException
e) {

            throw new JobPersistenceException("Couldn't
insert fired trigger: "

                    +
e.getMessage(), e);

        }

        Date
prevFireTime = trigger.getPreviousFireTime();

        //
call triggered - to update the trigger's next-fire-time state...

        trigger.triggered(cal);

        String
state = STATE_WAITING;

        boolean force
= true;

        if (job.isConcurrentExectionDisallowed())
{

            state
= STATE_BLOCKED;

            force
= false;

            try {

                getDelegate().updateTriggerStatesForJobFromOtherState(conn,
job.getKey(),

                        STATE_BLOCKED,
STATE_WAITING);

                getDelegate().updateTriggerStatesForJobFromOtherState(conn,
job.getKey(),

                        STATE_BLOCKED,
STATE_ACQUIRED);

                getDelegate().updateTriggerStatesForJobFromOtherState(conn,
job.getKey(),

                        STATE_PAUSED_BLOCKED,
STATE_PAUSED);

            } catch (SQLException
e) {

                throw new JobPersistenceException(

                        "Couldn't
update states of blocked triggers: "

                                +
e.getMessage(), e);

            }

        }

        if (trigger.getNextFireTime()
== null)
{

            state
= STATE_COMPLETE;

            force
= true;

        }

        storeTrigger(conn,
trigger, job, true,
state, force, false);

        job.getJobDataMap().clearDirtyFlag();

        return new TriggerFiredBundle(job,
trigger, cal, trigger.getKey().getGroup()

                .equals(Scheduler.DEFAULT_RECOVERY_GROUP), new Date(),
trigger

                .getPreviousFireTime(),
prevFireTime, trigger.getNextFireTime());

    }

qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));

shell.initialize(qs);

3.

select TRIGGER_ACCESS from QRTZ2_LOCKS for update

4.

SELECT TRIGGER_NAME,

TRIGGER_GROUP,

NEXT_FIRE_TIME,

PRIORITY

FROM QRTZ2_TRIGGERS

WHERE SCHEDULER_NAME
= 'CRMscheduler'

AND TRIGGER_STATE
= 'ACQUIRED'

AND NEXT_FIRE_TIME
<= '{timekey
30s latter}'

AND (
MISFIRE_INSTR = -1

OR (
MISFIRE_INSTR != -1

AND NEXT_FIRE_TIME
>= '{timekey
now}' )
)

ORDER BY NEXT_FIRE_TIME ASC,

PRIORITY DESC;

5.

SELECT *

FROM QRTZ2_JOB_DETAILS

WHERE SCHEDULER_NAME
= CRMscheduler

AND JOB_NAME
= ?

AND JOB_GROUP
= ?;

6.

UPDATE TQRTZ2_TRIGGERS

SET TRIGGER_STATE
= 'ACQUIRED'

WHERE SCHED_NAME
= 'CRMscheduler'

AND TRIGGER_NAME
= '{triggerName}'

AND TRIGGER_GROUP
= '{triggerGroup}'

AND TRIGGER_STATE
= 'waiting';

7.

INSERT INTO QRTZ2_FIRED_TRIGGERS

(SCHEDULER_NAME,

ENTRY_ID,

TRIGGER_NAME,

TRIGGER_GROUP,

INSTANCE_NAME,

FIRED_TIME,

SCHED_TIME,

STATE,

JOB_NAME,

JOB_GROUP,

IS_NONCONCURRENT,

REQUESTS_RECOVERY,

PRIORITY)

VALUES( 'CRMscheduler',
?, ?, ?, ?, ?

, ?);

8.

commit;

12.

select STAT_ACCESS from QRTZ2_LOCKS for update

13.

SELECT TRIGGER_STATE FROM QRTZ2_TRIGGERS WHERE SCHEDULER_NAME
= 'CRMscheduler' AND TRIGGER_NAME
= ?

AND TRIGGER_GROUP
= ?;

14.

SELECT TRIGGER_STATE

FROM QRTZ2_TRIGGERS

WHERE SCHEDULER_NAME
= 'CRMscheduler'

AND TRIGGER_NAME
= ?

AND TRIGGER_GROUP
= ?;

14.

SELECT *

FROM QRTZ2_JOB_DETAILS

WHERE SCHEDULER_NAME
= CRMscheduler

AND JOB_NAME
= ?

AND JOB_GROUP
= ?

;

15.

SELECT *

FROM QRTZ2_CALENDARS

WHERE SCHEDULER_NAME
= 'CRMscheduler'

AND CALENDAR_NAME
= ?;

16.

UPDATE QRTZ2_FIRED_TRIGGERS

SET INSTANCE_NAME
= ?,

FIRED_TIME
= ?,

SCHED_TIME
= ?

,

ENTRY_STATE
= ?,

JOB_NAME
= ?,

JOB_GROUP
= ?,

IS_NONCONCURRENT
= ?,

REQUESTS_RECOVERY
= ?

WHERE SCHEDULER_NAME
= 'CRMscheduler'

AND ENTRY_ID
= ?;

17.

UPDATE TQRTZ2_TRIGGERS

SET TRIGGER_STATE
= ?

WHERE SCHED_NAME
= 'CRMscheduler'

AND TRIGGER_NAME
= '{triggerName}'

AND TRIGGER_GROUP
= '{triggerGroup}'

AND TRIGGER_STATE
= ？;

18.

UPDATE QRTZ2_TRIGGERS

SET JOB_NAME
= ?,

JOB_GROUP
= ?

,

DESCRIPTION
= ?,

NEXT_FIRE_TIME
= ?,

PREV_FIRE_TIME
= ?,

TRIGGER_STATE
= ?,

TRIGGER_TYPE
= ?

,

START_TIME
= ?,

END_TIME
= ?,

CALENDAR_NAME
= ?,

MISFIRE_INSTRUCTION
= ?,

PRIORITY
= ?

,

JOB_DATAMAP
= ?

WHERE SCHEDULER_NAME
= SCHED_NAME_SUBST

AND TRIGGER_NAME
= ?

AND TRIGGER_GROUP
= ?;

19.

commit;