原理

15224.png

源码分析

15225.png
15223.png

  1. private static final int RUNNING    = -1 << COUNT_BITS;
  2. private static final int SHUTDOWN   =  0 << COUNT_BITS;
  3. private static final int STOP       =  1 << COUNT_BITS;
  4. private static final int TIDYING    =  2 << COUNT_BITS;
  5. private static final int TERMINATED =  3 << COUNT_BITS;

execute方法：

  1. int c = ctl.get();
  2. 1、判断当前的线程数是否小于corePoolSize如果是，
  3. 使用入参任务通过addWord方法创建一个新的线程，
  4. 如果能完成新线程创建exexute方法结束，成功提交任务；
  5. if (workerCountOf(c) < corePoolSize) {
  6.     if (addWorker(command, true))
  7.         return;
  8.     c = ctl.get();
  9. }
  10. 2、在第一步没有完成任务提交；状态为运行并且能成功加入任务到工作队列后，
  11. 再进行一次check，如果状态在任务加入队列后变为了非运行（有可能是在执行到这里线程池shutdown了）
  12. 非运行状态下当然是需要reject；
  13. 然后再判断当前线程数是否为0（有可能这个时候线程数变为了0），如是，新增一个线程；
  14. if (isRunning(c) && workQueue.offer(command)) {
  15.     int recheck = ctl.get();
  16.     if (! isRunning(recheck) && remove(command))
  17.         reject(command);
  18.     else if (workerCountOf(recheck) == 0)
  19.         addWorker(null, false); 判断当前工作线程池数是否为0  
  20.         如果是创建一个null任务，任务在堵塞队列存在了就会从队列中取出 这样做的意义是
  21.         保证线程池在running状态必须有一个任务在执行
  25. }
  26. 3、如果不能加入任务到工作队列，将尝试使用任务新增一个线程，如果失败，
  27. 则是线程池已经shutdown或者线程池已经达到饱和状态，所以reject;
  28. 从上面新增任务的execute方法也可以看出，拒绝策略不仅仅是在饱和状态下使用，
  29. 在线程池进入到关闭阶段同样需要使用到；
  30. 上面的几行代码还不能完全清楚这个新增任务的过程，
  31. 肯定还需要清楚addWorker方法才行：
  32. else if (!addWorker(command, false))
  33.     reject(command);

1、判断当前的线程数是否小于corePoolSize如果是，使用入参任务通过addWord方法创建一个新的线程，如果能完成新线程创建exexute方法结束，成功提交任务；
2、在第一步没有完成任务提交；状态为运行并且能成功加入任务到工作队列后，再进行一次check，如果状态在任务加入队列后变为了非运行（有可能是在执行到这里线程池shutdown了），非运行状态下当然是需要reject；然后再判断当前线程数是否为0（有可能这个时候线程数变为了0），如是，新增一个线程；
3、如果不能加入任务到工作队列，将尝试使用任务新增一个线程，如果失败，则是线程池已经shutdown或者线程池已经达到饱和状态，所以reject;
从上面新增任务的execute方法也可以看出，拒绝策略不仅仅是在饱和状态下使用，在线程池进入到关闭阶段同样需要使用到；

addWorker方法

  1. private boolean addWorker(Runnable firstTask, boolean core) {
  2.     retry: goto写法 用于重试
  3.     for (;;) { 
  4.         int c = ctl.get();
  5.         int rs = runStateOf(c);
  7.         // Check if queue empty only if necessary.
  8.         if (rs >= SHUTDOWN &&
  9.             ! (rs == SHUTDOWN &&
  10.                firstTask == null &&
  11.                ! workQueue.isEmpty()))
  12.                线程状态非运行并且非shutdown状态任务为空，队列非空就不能新增线程了
  14.             return false;
  16.         for (;;) {
  17.             int wc = workerCountOf(c);
  18.             if (wc >= CAPACITY ||
  19.                 wc >= (core ? corePoolSize : maximumPoolSize))
  20.                 当前现场大于等于最大值 
  21.                 等于核心线程数 非核心大于等于线程池数 说明达到了阈值 
  22.                 最大线程数 就不新增线程
  23.                 return false;
  24.             if (compareAndIncrementWorkerCount(c)) ctl+1 工作线程池数量+1 如果成功
  25.             就跳出死循环。
  26.             cas操作 如果为true 新增成功 退出
  27.                 break retry;
  28.             c = ctl.get();  // Re-read ctl
  29.             if (runStateOf(c) != rs)
  30.                 continue retry; 进来的状态和此时的状态发生改变 重头开始 重试 
  31.             // else CAS failed due to workerCount change; retry inner loop
  32.         }
  33.     }
  34.     上面主要是对ctl工作现场+1
  36.     boolean workerStarted = false;
  37.     boolean workerAdded = false;
  38.     Worker w = null;
  39.     try {
  40.         w = new Worker(firstTask); 内部类 封装了线程和任务 通过threadfactory创建线程
  42.         final Thread t = w.thread; 毎一个worker就是一个线程数
  43.         if (t != null) {
  44.             final ReentrantLock mainLock = this.mainLock;
  45.             mainLock.lock();
  46.             try {
  47.                 // Recheck while holding lock.
  48.                 // Back out on ThreadFactory failure or if
  49.                 // shut down before lock acquired.
  50.                 重新获取当前线程状态
  51.                 int rs = runStateOf(ctl.get());
  52.                  小于shutdown就是running状态
  53.                 if (rs < SHUTDOWN ||
  54.                     (rs == SHUTDOWN && firstTask == null)) {
  55.                         SHUTDOWN 和firstTask 为空是从队列中处理任务 那就可以放到集合中
  56.                        线程还没start 就是alive就直接异常
  57.                      if (t.isAlive()) // precheck that t is startable
  58.                         throw new IllegalThreadStateException();
  59.                     workers.add(w);
  60.                     int s = workers.size();
  61.                     if (s > largestPoolSize)
  62.                         largestPoolSize = s; 记录最大线程数
  63.                     workerAdded = true;
  64.                 }
  65.             } finally {
  66.                 mainLock.unlock();
  67.             }
  68.             if (workerAdded) {
  69.                 t.start(); 启动线程
  70.                 workerStarted = true;
  71.             }
  72.         }
  73.     } finally {
  74.         if (! workerStarted)
  75.             addWorkerFailed(w);//失败回退 从wokers移除w 线程数减1 尝试结束线程池
  76.     }
  77.     return workerStarted;
  1. private final class Worker
  2.     extends AbstractQueuedSynchronizer
  3.     implements Runnable
  4. {
  5.     /**
  6.      * This class will never be serialized, but we provide a
  7.      * serialVersionUID to suppress a javac warning.
  8.      */
  9.     private static final long serialVersionUID = 6138294804551838833L;
  11.     /** Thread this worker is running in.  Null if factory fails. */
  12.     正在运行woker线程
  13.     final Thread thread;
  14.     /** Initial task to run.  Possibly null. */
  15.     传入的任务
  16.     Runnable firstTask;
  17.     /** Per-thread task counter */
  18.     完成的任务数 监控用
  19.     volatile long completedTasks;
  21.     /**
  22.      * Creates with given first task and thread from ThreadFactory.
  23.      * @param firstTask the first task (null if none)
  24.      */
  25.     Worker(Runnable firstTask) {
  26.         禁止线程中断 
  27.         setState(-1); // inhibit interrupts until runWorker
  28.         this.firstTask = firstTask;
  29.         this.thread = getThreadFactory().newThread(this);
  30.     }
  32.     /** Delegates main run loop to outer runWorker  */
  33.     public void run() {
  34.         runWorker(this);

runwoker方法：

  1. final void runWorker(Worker w) {
  2.     Thread wt = Thread.currentThread();//获取当前线程
  3.     Runnable task = w.firstTask;
  4.     w.firstTask = null;
  5.     w.unlock(); // allow interrupts 把state从-1改为0 意思是可以允许中断
  6.     boolean completedAbruptly = true;
  7.     try { task不为空 或者阻塞队列中拿到了任务
  8.         while (task != null || (task = getTask()) != null) {
  9.             w.lock();
  10.             // If pool is stopping, ensure thread is interrupted;
  11.             // if not, ensure thread is not interrupted.  This
  12.             // requires a recheck in second case to deal with
  13.             // shutdownNow race while clearing interrupt
  14.             如果当前线程池状态等于stop 就中断
  15.             //Thread.interrupted() 中断标志
  16.             if ((runStateAtLeast(ctl.get(), STOP) ||
  17.                  (Thread.interrupted() &&
  18.                   runStateAtLeast(ctl.get(), STOP))) &&
  19.                 !wt.isInterrupted())
  20.                 wt.interrupt();
  21.             try {
  22.                 beforeExecute(wt, task);
  23.                 Throwable thrown = null;
  24.                 try {
  25.                     task.run();
  26.                 } catch (RuntimeException x) {
  27.                     thrown = x; throw x;
  28.                 } catch (Error x) {
  29.                     thrown = x; throw x;
  30.                 } catch (Throwable x) {
  31.                     thrown = x; throw new Error(x);
  32.                 } finally {
  33.                     afterExecute(task, thrown);
  34.                 }
  35.             } finally {
  36.                 task = null; 这设置为空 等下次循环就会从队列里面获取
  37.                 w.completedTasks++; 完成任务数+1
  38.                 w.unlock();
  39.             }
  40.         }
  41.         completedAbruptly = false;
  42.     } finally { 
  43.         processWorkerExit(w, completedAbruptly);
  44.     }

getTask方法：

  1. private Runnable getTask() {
  2.     boolean timedOut = false; // Did the last poll() time out?
  4.     for (;;) {
  5.         int c = ctl.get();
  6.         int rs = runStateOf(c);//获取线程池运行状态
  8.             shuitdown或者weikong 那就工作现场-1 同事返回为null 
  9.         // Check if queue empty only if necessary.
  10.         if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
  11.             decrementWorkerCount();
  12.             return null;
  13.         }
  14.              重新获取工作线程数
  15.         int wc = workerCountOf(c);
  16.         timed是标志超时销毁
  17.         allowCoreThreadTimeOut  true 核心线程池也是可以销毁的
  18.         // Are workers subject to culling?
  19.         boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  21.         if ((wc > maximumPoolSize || (timed && timedOut))
  22.             && (wc > 1 || workQueue.isEmpty())) {
  23.             if (compareAndDecrementWorkerCount(c))
  24.                 return null;
  25.             continue;
  26.         }
  28.         try {
  29.             Runnable r = timed ?
  30.                 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  31.                 workQueue.take();
  32.             if (r != null)
  33.                 return r;
  34.             timedOut = true;
  35.         } catch (InterruptedException retry) {
  36.             timedOut = false;
  37.         }
  38.     }

processWorkerExit方法：

  1. private void processWorkerExit(Worker w, boolean completedAbruptly) {
  2.     if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
  3.         decrementWorkerCount();
  5.     final ReentrantLock mainLock = this.mainLock;
  6.     mainLock.lock();
  7.     try {
  8.         completedTaskCount += w.completedTasks;
  9.         workers.remove(w);
  10.     } finally {
  11.         mainLock.unlock();
  12.     }
  14.     tryTerminate();
  16.     int c = ctl.get();
  17.     if (runStateLessThan(c, STOP)) {
  18.         if (!completedAbruptly) {
  19.             int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
  20.             if (min == 0 && ! workQueue.isEmpty())
  21.                 min = 1;
  22.             if (workerCountOf(c) >= min)
  23.                 return; // replacement not needed
  24.         }
  25.         addWorker(null, false);
  26.     }

拒绝策略

15206.png
ThreadPoolExecutor内部有实现4个拒绝策略：
(1)、CallerRunsPolicy，由调用execute方法提交任务的线程来执行这个任务；
(2)、AbortPolicy，抛出异常RejectedExecutionException拒绝提交任务；
(3)、DiscardPolicy，直接抛弃任务，不做任何处理；
(4)、DiscardOldestPolicy，去除任务队列中的第一个任务（最旧的），重新提交；

ScheduledThreadPoolExecutor

作用：定时线程池
schedule：延迟多长时间之后只执行一次；
scheduledAtFixedRate固定：延迟指定时间后执行一次，之后按照固定的时长周期执行；
scheduledWithFixedDelay非固定：延迟指定时间后执行一次，之后按照：上一次任务执行时长 + 周期的时长 的时间去周期执行；

  1. private void delayedExecute(RunnableScheduledFuture<?> task) {
  2.     //如果线程池不是RUNNING状态，则使用拒绝策略把提交任务拒绝掉
  3.     if (isShutdown())
  4.         reject(task);
  5.     else {
  6.         //与ThreadPoolExecutor不同，这里直接把任务加入延迟队列
  7.         super.getQueue().add(task);
  8.         //如果当前状态无法执行任务，则取消
  9.         if (isShutdown() &&
  10.             !canRunInCurrentRunState(task.isPeriodic()) &&
  11.             remove(task))
  12.             task.cancel(false);
  13.         else
  14.         //和ThreadPoolExecutor不一样，corePoolSize没有达到会增加Worker;
  15.         //增加Worker，确保提交的任务能够被执行
  16.             ensurePrestart();
  17.     }

offer方法：

  1. public boolean offer(Runnable x) {
  2.     if (x == null)
  3.         throw new NullPointerException();
  4.     RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
  5.     final ReentrantLock lock = this.lock;
  6.     lock.lock();
  7.     try {
  8.         int i = size;
  9.         if (i >= queue.length)
  10.             // 容量扩增50%。
  11.             grow();
  12.         size = i + 1;
  13.         // 第一个元素,其实这里也可以统一进行sift-up操作,没必要特判。
  14.         if (i == 0) {
  15.             queue[0] = e;
  16.             setIndex(e, 0);
  17.         } else {
  18.             // 插入堆尾。
  19.             siftUp(i, e);
  20.         }
  21.         // 如果新加入的元素成为了堆顶,则原先的leader就无效了。
  22.         if (queue[0] == e) {
  23.             leader = null;
  24.             // 由于原先leader已经无效被设置为null了,这里随便唤醒一个线程(未必是原先的leader)来取走堆顶任务。
  25.             available.signal();
  26.         }
  27.     } finally {
  28.         lock.unlock();
  29.     }
  30.     return true;

siftup方法：

  1. private void siftUp(int k, RunnableScheduledFuture<?> key) {
  2.     // 找到父节点的索引
  3.     while (k > 0) {
  4.         // 获取父节点
  5.         int parent = (k ­- 1) >>> 1;
  6.         RunnableScheduledFuture<?> e = queue[parent];
  7.         // 如果key节点的执行时间大于父节点的执行时间，不需要再排序了
  8.         if (key.compareTo(e) >= 0)
  9.             break;
  10.         // 如果key.compareTo(e) < 0，
  11.         说明key节点的执行时间小于父节点的执行时间，需要把父节点移到后面
  12.         queue[k] = e;
  13.         setIndex(e, k);
  14.         // 设置索引为k
  15.         k = parent;
  16.     }
  17.     // key设置为排序后的位置中
  18.     queue[k] = key;
  19.     setIndex(key, k);

任务执行:

  1. public void run() {
  2.     // 是否周期性，就是判断period是否为0。
  3.     boolean periodic = isPeriodic();
  4.     // 检查任务是否可以被执行。
  5.     if (!canRunInCurrentRunState(periodic))
  6.         cancel(false);
  7.     // 如果非周期性任务直接调用run运行即可。
  8.     else if (!periodic)
  9.         ScheduledFutureTask.super.run();
  10.     // 如果成功runAndRest，则设置下次运行时间并调用reExecutePeriodic。
  11.     else if (ScheduledFutureTask.super.runAndReset()) {
  12.         setNextRunTime();
  13.         // 需要重新将任务(outerTask)放到工作队列中。此方法源码会在后文介绍ScheduledThreadPoolExecutor本身API时提及。
  14.         reExecutePeriodic(outerTask);
  15.     }