public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory) {
           super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,new DelayedWorkQueue(), threadFactory);
}

private void runPeriodic() {
          boolean ok = ScheduledFutureTask.super.runAndReset();
          boolean down = isShutdown();
          // Reschedule if not cancelled and not shutdown or policy allows
      if (ok && (!down ||(getContinueExistingPeriodicTasksAfterShutdownPolicy() && !isStopped()))) {
                long p = period;
                if (p > 0)
                      time += p;
                else
                      time = triggerTime(-p);
     
                ScheduledThreadPoolExecutor.super.getQueue().add(this);
         }
         // This might have been the final executed delayed
        // task.  Wake up threads to check.
        else if (down)
              interruptIdleWorkers();
}

2.       CompletionService

CompletionService定義了線程池執行任務集,可以依次拿到任務執行完畢的Future,ExecutorCompletionService是其實現類,先舉個例子,如下代碼,這個例子中,需要注意ThreadPoolExecutor核心池一定保證能夠讓任務提交并且馬上執行,而不是放到等待隊列中去,那樣次序將會無法控制,CompletionService也將失去效果(其實核心池中的任務完成順序還是準確的).

public static void main(String[] args) throws InterruptedException, ExecutionException{
    ThreadPoolExecutor es=new ThreadPoolExecutor(10, 15, 2000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10),new ThreadPoolExecutor.AbortPolicy());
    CompletionService<String> cs=new ExecutorCompletionService<String>(es);    
    cs.submit(new Callable<String>() {
     @Override
     public String call() throws Exception {
         Thread.currentThread().sleep(1000);
         return "i am sleeped 1000 milliseconds";
     }
    });
        
    cs.submit(new Callable<String>() {
     @Override
     public String call() throws Exception {
         Thread.currentThread().sleep(5000);
         return "i am sleeped 5000 milliseconds";
     }
    });
        
    cs.submit(new Callable<String>() {
     @Override
     public String call() throws Exception {
         Thread.currentThread().sleep(4000);
         return "i am sleeped 4000 milliseconds";
     }
    });
        
    cs.submit(new Callable<String>() {
     @Override
         public String call() throws Exception {
          Thread.currentThread().sleep(2000);
              return "i am sleeped 2000 milliseconds";
      }
});
        
    for(int i=0;i<4;i++){
        Future<String> fu=cs.take();
        System.out.println(fu.get());
    }
}

ExecutorCompletionService并不復雜,關鍵的一個點就是它的內部類QueueingFuture繼承了FutureTask類,并且實現了done()方法,done()方法是在線程池任務執行完畢,最后調用FutureTask的方法(這在 JAVA LOCK代碼淺析(http://m.tkk7.com/BucketLi/archive/2010/09/30/333471.html)一文中對于FutureTask代碼解析有提到)

QueueingFuture的done()方法實現是將執行完的任務(FutureTask)丟入全局的完成隊列中(completionQueue),那么take是從這個blockingqueue中取元素.也就是任務完成就會有元素,即生產者消費者.

這種實現的思想是將原本在單個FutureTask上的等待轉化為在BlockingQueue上的等待,即對全部FutureTask的等待,從而達到哪個先完成,哪個就可取執行結果的效果.

private class QueueingFuture extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task) {
        super(task, null);
        his.task = task;
    }
    protected void done() { completionQueue.add(task); }
    private final Future<V> task;
}

總結:
JUC提供的線程池體系核心是在ThreadPoolExecutor, 而ScheduledThreadPoolExecutor和ExecutorCompletionService只是對其擴展,這里沒有去細講Executors這個便捷類,這個類提供很多便捷的線程池構建方法.各位使用的時候不妨去看下.