clDream quartz的初步总结及配置优化

1.scheduler 1. Scheduler就是Quartz的大脑,所有任务都是由它来设施。 Scheduler包含一个两个重要组件: JobStore和ThreadPool。 JobStore是会来存储运行时信息的,包括Job、JobDetail、Trigger以及业务锁等。它有多种实现RAMJob(内存实现),JobStoreTX(JDBC,事务由Quartz管理),JobStoreCMT(JDBC,使用容器事务),ClusteredJobStore(集群实现)。 ThreadPool就是线程池,Quartz有自己的线程池实现。所有任务的都会由线程池执行。 2.SchdulerFactory,顾名思义就是来用创建Schduler了,有两个实现:DirectSchedulerFactory和 StdSchdulerFactory。前者可以用来在代码里定制你自己的Schduler参数。后者是直接读取classpath下的quartz.properties(不存在就都使用默认值)配置来实例化Schduler。通常来讲,我们使用StdSchdulerFactory也就足够了。 1 2 3 4 5 6 org.quartz.scheduler.instanceName = DefaultQuartzScheduler org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool org.quartz.threadPool.threadCount = 10 org.quartz.threadPool.threadPriority = 5 org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread = true org.quartz.jobStore.class = org.quartz.simpl.RAMJobStore    2.Trigger 2.1 simpleTrigger SimpleTrigger可以满足的调度需求是:在具体的时间点执行一次,或者在具体的时间点执行,并且以指定的间隔重复执行若干次。比如,你有一个trigger,你可以设置它在2018年12月9日的上午11:23:54准时触发,或者在这个时间点触发,并且每隔2秒触发一次,一共重复5次。 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 public static void main(String[] args) throws SchedulerException { // 创建Scheduler SchedulerFactory sf = new StdSchedulerFactory(); Scheduler scheduler = sf.getScheduler(); // 需求:我要在5秒之后执行任务,时间间隔为2秒,最多执行100次 long currentTime = System.currentTimeMillis(); long delayTime = currentTime + 5 * 1000l; // 5秒之后执行任务 // 设置Trigger(不再使用静态方法) Trigger trigger = TriggerBuilder.newTrigger() // 使用TriggerBuilder创建Trigger .withIdentity("trigger1", "group1") .startAt(new Date(delayTime)) .withSchedule(SimpleScheduleBuilder .simpleSchedule() // 使用SimpleScheduleBuilder创建simpleSchedule .withIntervalInSeconds(2) // 时间间隔为2秒 .withRepeatCount(99)) // 最多执行100次,此处需要注意,不包括第一次执行的 .build(); // 设置JobDetail(不再使用静态方法) JobDetail jobDetail = JobBuilder.newJob((MyJobDetail.class)) // 使用JobBuilder创建JobDetail .withIdentity("jobDetail1", "group1") .usingJobData("user", "AlanShelby") .build(); // 设置scheduler scheduler.scheduleJob(jobDetail, trigger); // 启动、停止Scheduler scheduler.start(); try { Thread.sleep(500000); } catch (InterruptedException e) { e.printStackTrace(); } scheduler.shutdown(); }   2.2 cronTrigger 1、适用于更为复杂的需求,它类似于Linux系统中的crontab,但要比crontab更强大。它基本上覆盖了之前章节讲到的三个类型的功能(并不是全部功能),相对于前三个类型,cronTrigger也更难理解。 2、它适合的任务类似于:每天0:00,9:00,18:00各执行一次。 3、它的属性只有一个Cron表达式,下面有对cron表达式详细的讲解。 1 2 3 4 Trigger trigger = TriggerBuilder.newTrigger() .withIdentity("trigger1", "group1") .withSchedule(CronScheduleBuilder.cronSchedule("*/5 * * * * ?")) // 每5秒钟执行一次 .build();    3.job Job是有可能并发执行的,比如一个任务要执行10秒中,而调度算法是每秒中触发1次,那么就有可能多个任务被并发执行。 有时候我们并不想任务并发执行,比如这个任务要去”获得数据库中所有未发送邮件的名单”,如果是并发执行,就需要一个数据库锁去避免一个数据被多次处理。这个时候一个@DisallowConcurrentExecution解决这个问题。 1 2 3 4 5 6 @DisallowConcurrentExecution public class DoNothingJob implements Job { public void execute(JobExecutionContext context) throws JobExecutionException { System.out.println("do nothing"); } }    4.调度核心类QuartzSchedulerThread 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 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 /** *

* The main processing loop of the QuartzSchedulerThread. *

*/ @Override public void run() { int acquiresFailed = 0; 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) { } // reset failure counter when paused, so that we don't // wait again after unpausing acquiresFailed = 0; } if (halted.get()) { break; } } // wait a bit, if reading from job store is consistently // failing (e.g. DB is down or restarting).. if (acquiresFailed > 1) { try { long delay = computeDelayForRepeatedErrors(qsRsrcs.getJobStore(), acquiresFailed); Thread.sleep(delay); } catch (Exception ignore) { } } int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads(); if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads... List triggers; long now = System.currentTimeMillis(); clearSignaledSchedulingChange(); 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()) { now = System.currentTimeMillis(); long triggerTime = triggers.get(0).getNextFireTime().getTime(); long timeUntilTrigger = triggerTime - now; while(timeUntilTrigger > 2) { synchronized (sigLock) { if (halted.get()) { break; } 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) { } } } if(releaseIfScheduleChangedSignificantly(triggers, triggerTime)) { break; } now = System.currentTimeMillis(); timeUntilTrigger = triggerTime - now; } // this happens if releaseIfScheduleChangedSignificantly decided to release triggers if(triggers.isEmpty()) continue; // set triggers to 'executing' List bndles = new ArrayList(); boolean goAhead = true; synchronized(sigLock) { goAhead = !halted.get(); } if(goAhead) { try { List 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; } } 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) 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) { // 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) } long now = System.currentTimeMillis(); long waitTime = now + getRandomizedIdleWaitTime(); long timeUntilContinue = waitTime - now; synchronized(sigLock) { try {
50000+
5万行代码练就真实本领
17年
创办于2008年老牌培训机构
1000+
合作企业
98%
就业率

联系我们

电话咨询

0532-85025005

扫码添加微信