quartz定时任务框架调度机制解析

转自集群调度机制调研及源码分析

quartz2.2.1集群调度机制调研及源码分析

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

我们看到初始化一个调度器需要用工厂类获取实例:

然后启动:

下面跟进StdSchedulerFactory的getScheduler()方法:

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

当调用sch.start();方法时,scheduler做了如下工作:

1.通知listener开始启动

2.启动调度器线程

3.启动plugin

4.通知listener启动完成

调度器每次获取到的trigger是30s内需要执行的,所以要等待一段时间至trigger执行前2ms.在等待过程中涉及到一个新加进来更紧急的trigger的处理逻辑.分析写在注释中,不再赘述.

可以看到调度器的只要在运行状态,就会不停地执行调度流程.值得注意的是,在流程的最后线程会等待一个随机的时间.这就是quartz自带的负载平衡机制.

以下是三个步骤的跟进:

触发器的获取

调度器调用:

在数据库中查找一定时间范围内将会被触发的trigger.参数的意义如下:参数1:nolaterthan = now+3000ms,即未来30s内将会被触发.参数2 最大获取数量,大小取线程池线程剩余量与定义值得较小者.参数3 时间窗口 默认为0,程序会在nolaterthan后加上窗口大小来选择trigger.quratz会在每次触发trigger后计算出trigger下次要执行的时间,并在数据库QRTZ2_TRIGGERS中的NEXT_FIRE_TIME字段中记录.查找时将当前毫秒数与该字段比较,就能找出下一段时间内将会触发的触发器.查找时,调用在JobStoreSupport类中的方法:

进一步解释:quratz在获取数据库资源之前,先要以for update方式访问LOCKS表中相应LOCK_NAME数据将改行锁定.如果在此前该行已经被锁定,那么等待,如果没有被锁定,那么读取满足要求的trigger,并把它们的status置为STATE_ACQUIRED,如果有tirgger已被置为STATE_ACQUIRED,那么说明该trigger已被别的调度器实例认领,无需再次认领,调度器会忽略此trigger.调度器实例之间的间接通信就体现在这里.

JobStoreSupport.acquireNextTrigger()方法中:

int rowsUpdated = getDelegate().updateTriggerStateFromOtherState(conn, triggerKey, STATE_ACQUIRED, STATE_WAITING);

最后释放锁,这时如果下一个调度器在排队获取trigger的话,则仍会执行相同的步骤.这种机制保证了trigger不会被重复获取.按照这种算法正常运行状态下调度器每次读取的trigger中会有相当一部分已被标记为被获取.

获取trigger的过程进行完毕.

触发trigger:

QuartzSchedulerThread line336:

List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);

调用JobStoreSupport类的triggersFired()方法:

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

该方法做了以下工作:

1.获取trigger当前状态

2.通过trigger中的JobKey读取trigger包含的Job信息

3.将trigger更新至触发状态

4.结合calendar的信息触发trigger,涉及多次状态更新

5.更新数据库中trigger的信息,包括更改状态至STATE_COMPLETE,及计算下一次触发时间.

6.返回trigger触发结果的数据传输类TriggerFiredBundle

从该方法返回后,trigger的执行过程已基本完毕.回到执行quratz操作规范的executeInNonManagedTXLock方法,将数据库锁释放.

trigger触发操作完成

Job执行过程:

再回到线程类QuartzSchedulerThread的 line353这时触发器都已出发完毕,job的详细信息都已就位

QuartzSchedulerThread line:368

为每个Job生成一个可运行的RunShell,并放入线程池运行.

在最后调度线程生成了一个随机的等待时间,进入短暂的等待,这使得其他节点的调度器都有机会获取数据库资源.如此就实现了quratz的负载平衡.

这样一次完整的调度过程就结束了.调度器线程进入下一次循环.

总结:

简单地说,quartz的分布式调度策略是以数据库为边界资源的一种异步策略.各个调度器都遵守一个基于数据库锁的操作规则保证了操作的唯一性.同时多个节点的异步运行保证了服务的可靠.但这种策略有自己的局限性.摘录官方文档中对quratz集群特性的说明:

Only one node will fire the job for each firing. What I mean by that is, if the job has a repeating trigger that tells it to fire every 10 seconds, then at 12:00:00 exactly one node will run the job, and at 12:00:10 exactly one node will run the job, etc. It won't necessarily be the same node each time - it will more or less be random which node runs it. The load balancing mechanism is near-random for busy schedulers (lots of triggers) but favors the same node for non-busy (e.g. few triggers) schedulers. 

The clustering feature works best for scaling out long-running and/or cpu-intensive jobs (distributing the work-load over multiple nodes). If you need to scale out to support thousands of short-running (e.g 1 second) jobs, consider partitioning the set of jobs by using multiple distinct schedulers (including multiple clustered schedulers for HA). The scheduler makes use of a cluster-wide lock, a pattern that degrades performance as you add more nodes (when going beyond about three nodes - depending upon your database's capabilities, etc.).

说明指出,集群特性对于高cpu使用率的任务效果很好,但是对于大量的短任务,各个节点都会抢占数据库锁,这样就出现大量的线程等待资源.这种情况随着节点的增加会越来越严重.

附：

通讯图中关键步骤的主要sql语句：

海外公司注册、海外银行开户、跨境平台代入驻、VAT、EPR等知识和在线办理：https://www.xlkjsw.com

原标题：quartz定时任务框架调度机制解析

关键词：