java中的scheduledThreadPool

今天看了一下jdk的定时任务的源码，将此记录再次一下我的感受，方便出问题时的查找。

这次突发想看定时任务的原理的原因是在代码中使用到了，以为我们要定期的执行一个任务，但是我又不想使用线程+while true的操作，所以就想起了jdk的功能，但是有知道他的原理，害怕出岔子，所以就看了一下。看这个博客需要知道线程池的原理，也就是Executors.newFixedThreadPool 或者 是Executors.newCachedThreadPool的实现原理。（其实很简单，就是使用了一个阻塞队列，然后启动多个线程去队列中获取任务，如果队列中任务没有，就阻塞线程，线程的启动数量是受到两个参数的控制的，就这么简单）。

我是使用了Executors.newSingleThreadScheduledExecutor，不过也有这个方法：Executors.newScheduledThreadPool，两者的区别是前者只能启动一个线程，而后者可以启动多个(自己制定个数）。在方法里面最终都是调用的

 public ScheduledThreadPoolExecutor(int corePoolSize) {//参数表示启动的线程的最大数量，
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,//不要关心第二个参数，这里和newFixedThreadPool是不一样的，不会因为第二个参数大于第一个参数就会生成超过corePoolSize数量的线程。
              new DelayedWorkQueue());
    }

 这里的super即是ThreadPoolExecutor，说明定时调度的线程池和非固定调度的线程池都是继承自ThreadPoolExecutor，不过传入的第二个参数不会起作用（在fixexThreadPool或者CacheThreadPool中会起作用的），还有最关键的一个是最后的阻塞队列，在FixedThreadPool（或者CachedThreaPool）中，都是使用的LinkedBlockingQueue，但是这里传入的是DelayedWordQueue，我们先介绍一下这个类吧。

这个类也是一个阻塞队列，他实现了BlockingQueue的接口，但是他里面使用的容器是一个堆（用数组实现的），而不是像之前的LinkedBlockingQueue中使用链表实现。他的原理很简单，当读取的时候，如果堆中没有任何的内容，则会使用lock condition进行阻塞，如果有则使用lock锁定，然后从队列中读取，然后解锁；当添加的时候，也会使用lock进行锁定，添加完成后唤醒之前沉睡的线程（如果有的话），至于那个堆，一定要制定排序的逻辑才可以，我们看一下他的offer方法（即队列的添加方法），他会将加入的每一个Runnable转化为RunnableScheduledFuture（看来这个类的使用有很大的限制），我们看一下这个RunnableScheduledFuture类，他既是一个Future，也是一个Runnable，还是一个Delay，Delay第一次遇到，他很简单，其实就是一个可以调用getDelay方法的东西，DelayedWordQueue中根据这个Delay的getDelay方法对加入的Runnable记性排序，delay小的排在前面，他的真正含义是，对于要定期执行的任务（也就是RunnableScheduledFuture）进行排序，每一个线程从队列中获得任务时将会先获得堆顶的任务，这样就保证了离开始最近的任务有限被调用。

 

上面看完了DelayedWordQueue，也就是最终使用的容器，我们再看一下真正的调度吧，我们看一下ScheduledThreadPoolExecutor这个类的scheduleAtFixedRate，他表示

    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (period <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),unit.toNanos(period));//用ScheduledFutureTask封装runable，
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);//将scheduledFuture封装的任务添加到对列中，记住，这样的话，在线程从队列中获得任务，执行run方法的时候，就会是调用的ScheduledFutureTask的run方法了
        return t;
}

 

我们再看一下scheduleWithFixedDelay，可以发现，他的操作几乎和上面的fixedRate一样

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (delay <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(-delay));//唯一不同的地方就是将delay变为负数了
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

 记住了最后线程从队列中获得的任务是被ScheduledFutureTask，我们看一下他的run方法，

public void run() {
            boolean periodic = isPeriodic();//这个的依据是  periorid是不是0，当我们调用schedule方法的时候，就是传入的0，fixedRate或者fixDelay不是0
            if (!canRunInCurrentRunState(periodic))
                cancel(false);
            else if (!periodic)
                ScheduledFutureTask.super.run();//从这里看出，如果是调用schedule方法，是不会重复的
            else if (ScheduledFutureTask.super.runAndReset()) {//如果是重复的，即period不是0，就会进入super的runAndReset方法，即执行run方法，如果返回的是true，然后进入下面，重点来了！
                setNextRunTime();//设置下次的执行时间（从这里看出是重复执行的）
                reExecutePeriodic(outerTask);//重复执行
            }
        }

 我们先看一下ScheduledFutureTask.super.runAndReset()方法，即什么时候会定期实行：

 protected boolean runAndReset() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return false;
        boolean ran = false;
        int s = state;
        try {
            Callable<V> c = callable;
            if (c != null && s == NEW) {
                try {
                    c.call(); // don't set result，看这里，如果出错了，就会ran=false
                    ran = true;
                } catch (Throwable ex) {
                    setException(ex);
                }
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
        return ran && s == NEW;//如果ran=false，就会返回false，即不会重复任务
    }

 从上面的代码可以看出，如果在执行任务的时候报错了，就不会重复执行这个任务。

我们再看一下setNextRuntTime

private void setNextRunTime() {
      long p = period;
      if (p > 0)//如果是fixedRate，就会是将上一次的执行时间，加上间隔时间，
         time += p;
      else
         time = triggerTime(-p);//如果是fixedDelay是将现在的时间（绝对时间）+ delay
}

 从上面的方法可以判断出，fixedDelay是在任务执行后的当前时间上之后delay时间再执行任务。如果是fixedRate，是以上一次任务的应该执行时间+delay作为下一次的执行时间，这个时间可能在当前的绝对时间之前！

over!