zoukankan      html  css  js  c++  java
  • 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也就足够了。

    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次。

    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表达式详细的讲解。

    Trigger trigger = TriggerBuilder.newTrigger()
    	.withIdentity("trigger1", "group1")
    	.withSchedule(CronScheduleBuilder.cronSchedule("*/5 * * * * ?")) // 每5秒钟执行一次
    	.build();
    

      

    3.job

    Job是有可能并发执行的,比如一个任务要执行10秒中,而调度算法是每秒中触发1次,那么就有可能多个任务被并发执行。

    有时候我们并不想任务并发执行,比如这个任务要去”获得数据库中所有未发送邮件的名单”,如果是并发执行,就需要一个数据库锁去避免一个数据被多次处理。这个时候一个@DisallowConcurrentExecution解决这个问题。

    @DisallowConcurrentExecution 
    public class DoNothingJob implements Job {
        public void execute(JobExecutionContext context) throws JobExecutionException {
            System.out.println("do nothing");
        }
    }
    

      

    4.调度核心类QuartzSchedulerThread

    /**
         * <p>
         * The main processing loop of the <code>QuartzSchedulerThread</code>.
         * </p>
         */
        @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<OperableTrigger> 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<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>();
    
                            boolean goAhead = true;
                            synchronized(sigLock) {
                                goAhead = !halted.get();
                            }
                            if(goAhead) {
                                try {
                                    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;
                                }
    
                            }
    
                            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 {
                          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;
        }
    

      

    4.1 blockForAvailableThreads

    就是qsRsrcs.getThreadPool().blockForAvailableThreads(),如果线程池满了的话,则会阻塞,因而会影响调度的准确性。
    
    int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
    
    获取可用的线程数量。通常在第一次时候这个数量等于配置中配置的参数:
    
    org.quartz.threadPool.threadCount = 20
    

      

    4.2 maxBatchSize

    triggers = qsRsrcs.getJobStore().acquireNextTriggers(
                                    now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
    

    这个参数的意思是批量查询的数量,但并不是你配置多少它每次就能查询多少,这算一个优化的配置项,因为在jdbc store的时候,减少对数据库的轮询次数算是一个比较大的优化;Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow())可以看到这里会取配置的批量的数和可用线程的最小数,所以批量数可以配置成和线程数大小一致:

    org.quartz.threadPool.threadCount= 20
    org.quartz.scheduler.batchTriggerAcquisitionMaxCount= 20  

      

  • 相关阅读:
    Vue表单输入绑定(文本框和复选框)
    Vue登录方式的切换
    IDEA导包(以junit为例)
    反射相关类
    dom4j加载xml文件
    Vue购物车
    IDEA基本设置和快捷键大全
    react组件间的通信-父子通信,子父通信,兄弟通信
    react受控组件
    react中数据承载--props和state用法
  • 原文地址:https://www.cnblogs.com/clovejava/p/10093257.html
Copyright © 2011-2022 走看看