Java线程池实现原理和源码分析

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

/**
 * 任务阻塞队列 
 */
private final BlockingQueue<Runnable> workQueue; 
/**
 * 非公平的互斥锁（可重入锁）
 */
private final ReentrantLock mainLock = new ReentrantLock();
/**
 * 线程集合一个Worker对应一个线程，没有核心线程的说话，只有核心线程数
 */
private final HashSet<Worker> workers = new HashSet<Worker>();
/**
 * 配合mainLock通过Condition能够更加精细的控制多线程的休眠与唤醒
 */
private final Condition termination = mainLock.newCondition();
/**
 * 线程池中线程数量曾经达到过的最大值。
 */
private int largestPoolSize;  
/**
 * 已完成任务数量
 */
private long completedTaskCount;
/**
 * ThreadFactory对象，用于创建线程。
 */
private volatile ThreadFactory threadFactory;  
/**
 * 拒绝策略的处理句柄
 * 现在默认提供了CallerRunsPolicy、AbortPolicy、DiscardOldestPolicy、DiscardPolicy
 */
private volatile RejectedExecutionHandler handler;
/**
 * 线程池维护线程（超过核心线程数）所允许的空闲时间
 */
private volatile long keepAliveTime;
/**
 * 允许线程池中的核心线程超时进行销毁
 */
private volatile boolean allowCoreThreadTimeOut;  
/**
 * 线程池维护线程的最小数量，哪怕是空闲的  
 */
private volatile int corePoolSize;
/**
 * 线程池维护的最大线程数量，线程数超过这个数量之后新提交的任务就需要进入阻塞队列
 */
private volatile int maximumPoolSize;

public static ExecutorService newCachedThreadPool() {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                  60L, TimeUnit.SECONDS,
                  new SynchronousQueue<Runnable>());
}

public static ExecutorService newSingleThreadExecutor() {
  return new FinalizableDelegatedExecutorService
    (new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
  return new ThreadPoolExecutor(nThreads, nThreads,
                  0L, TimeUnit.MILLISECONDS,
                  new LinkedBlockingQueue<Runnable>(),
                  threadFactory);
}

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

pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());

pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());

public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

public static class AbortPolicy implements RejectedExecutionHandler {
  public AbortPolicy() { }
  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +
                                         " rejected from " +
                                         e.toString());
  }
}

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public CallerRunsPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

public static class DiscardPolicy implements RejectedExecutionHandler {
    public DiscardPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public DiscardOldestPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

private static final int COUNT_BITS = Integer.SIZE - 3;//32-3
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;//低29位都为1，高位都为0

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;//111
private static final int SHUTDOWN   =  0 << COUNT_BITS;//000
private static final int STOP       =  1 << COUNT_BITS;//001
private static final int TIDYING    =  2 << COUNT_BITS;//010
private static final int TERMINATED =  3 << COUNT_BITS;//011

// Packing and unpacking ctl
//计算当前运行状态，取高三位
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//计算当前线程数量，取低29位
private static int workerCountOf(int c)  { return c & CAPACITY; }
//通过状态和线程数生成ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

//AbstractExecutorService.java
public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}
//ThreadPoolExecutor.java
public void execute(Runnable command) {
  if (command == null)
    throw new NullPointerException();
  int c = ctl.get();//获取ctl
  //检查当前核心线程数，是否小于核心线程数的大小限制
  if (workerCountOf(c) < corePoolSize) {
    //没有达到核心线程数的大小限制，那么添家核心线程执行该任务
    if (addWorker(command, true))
      return;
    //如果添加失败，刷新ctl值
    c = ctl.get();
  }
  //再次检查线程池的运行状态，将任务添加到等待队列中
  if (isRunning(c) && workQueue.offer(command)) {
    int recheck = ctl.get();//刷新ctl值
    //如果当前线程池的装不是运行状态，那么移除刚才添加的任务
    if (! isRunning(recheck) && remove(command))
      reject(command);//移除成功后，使用拒绝策略处理该任务；
    else if (workerCountOf(recheck) == 0)//当前工作线程数为0
      //线程池正在运行，或者移除任务失败。
      //添加一个非核心线程，并不指定该线程的运行任务。
      //等线程创建完成之后，会从等待队列中获取任务执行。
      addWorker(null, false);
  } 
  //逻辑到这里说明线程池已经不是RUNNING状态，或者等待队列已满，需要创建一个新的非核心线程执行该任务；
  //如果创建失败，那么非核心线程已满，使用拒绝策略处理该任务；
  else if (!addWorker(command, false))
    reject(command);
}

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
  Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;//初始化的任务，可以为null
    this.thread = getThreadFactory().newThread(this);//Worker持有的线程
  }
  /**部分代码省略*/
	public void run() {
  	runWorker(this);
	}
}

private boolean addWorker(Runnable firstTask, boolean core) {
    retry://break和continue的跳出标签
    for (;;) {
        int c = ctl.get();//获取ctl的值
        int rs = runStateOf(c);//获取当前线程池的状态；
        /**
         * 1、如果当前的线程池状态不是RUNNING
         * 2、当前线程池是RUNNING而且没有添加新任务，而且等待队列不为空。这种情况下是需要创建执行线程的。
         * 所以满足1，但不满足2就创建执行线程失败，返回false。
         */
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
                firstTask == null &&
                ! workQueue.isEmpty()))
            return false;
        /**进入内层循环 */
        for (;;) {
            int wc = workerCountOf(c);//获取当前执行线程的数量
            /**
             * 1、工作线程数量大于或等于计数器的最大阈值，那么创建执行线程失败，返回false。
             * 2、如果当前创建的核心线程，那么工作线程数大于corePoolSize的话，创建执行线程失败，返回false。
             * 3、如果当前创建的是非核心线程，那么工作线程数大于maximumPoolSize的话，创建执行线程失败，返回false。
             */
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            //用CAS操作让线程数加1，如果成功跳出整个循环
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)//线程状态前后不一样，重新执行外循环
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
            //如果CAS操作由于工作线程数的增加失败，那么重新进行内循环
        }
    }
    /**就现在，线程数已经增加了。但是真正的线程对象还没有创建出来。*/
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();//加锁
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());
                /**
                 * 再次检查线程池的运行状态
                 * 1、如果是RUNNING状态，那么可以创建；
                 * 2、如果是SHUTDOWN状态，但没有执行线程，可以创建（创建后执行等待队列中的任务）
                 */
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    //检测该线程是否已经开启
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);//将改工作线程添加到集合中
                    int s = workers.size();
                    if (s > largestPoolSize)//更新线程池的运行时的最大线程数
                        largestPoolSize = s;
                    workerAdded = true;//标识工作线程添加成功
                }
            } finally {//释放锁
                mainLock.unlock();
            }
            if (workerAdded) {//如果工作线程添加成功，那么开启线程
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        //如果工作线程添加失败，那么进行失败处理
        //将已经增加的线程数减少，将添加到集合的工作线程删除
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    //解锁，允许中断
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //如果当前的工作线程已经有执行任务，或者可以从等待队列中获取到执行任务
        //getTask获取任务时候会进行阻塞
        while (task != null || (task = getTask()) != null) {
            w.lock();//开始执行，上锁
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            //判断线程是否需要中断
            //如果线程池状态是否为STOP\TIDYING\TERMINATED,同时当前线程没有被中断那么将当前线程进行中断
            if ((runStateAtLeast(ctl.get(), STOP) ||
                    (Thread.interrupted() &&
                    runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;//异常处理
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;//工作线程的当前任务置空
                w.completedTasks++;//当前工作线程执行完成的线程数+1
                w.unlock();//执行完成解锁
            }
        }
        completedAbruptly = false;//完成了所有任务，正常退出
    } finally {//执行工作线程的退出操作
        processWorkerExit(w, completedAbruptly);
    }
}

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {
        int c = ctl.get();//获取ctl的值
        int rs = runStateOf(c);//获取线程池状态

        // Check if queue empty only if necessary.
        /**
         * 1、rs为STOP\TIDYING\TERMINATED，标识无法继续执行任务
         * 2、等待队列中没有任务可以被执行
         * 工作线程数量减一
         */
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);//获取工作线程数量
        // Are workers subject to culling?
        //如果允许核心线程超时，或者当前工作线程数量大于核心线程数量。标识需要进行超时检测
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        /**
         * 1、如果当前工作线程数是否大于线程池可允许的最大工作线程数（maximumPoolSize可以动态设置）
         * ，或者当前需要进行超时控制并且上次从等待队列中获取执行任务发生了超时。
         * 2、如果当前不是唯一的线程，并且等待队列中没有需要执行的任务。
         * 这两种情况下一起存在就表示，工作线程发生了超时需要回收，所以对线程数进行-1；
         */
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))//线程数量减少成功，否则重新执行本次循环
                return null;
            continue;
        }
        try {
            //如果设置有超时，那么设定超时时间。否则进行无限的阻塞等待执行任务
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;//获取超时,设置标记
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    //completedAbruptly为true，标识该工作线程执行出现了异常，将工作线程数减一
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();
    //否则标识该工作线程为正常结束，这种情况下getTask方法中已经对工作线程进行了减一
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();//加锁
    try {
        completedTaskCount += w.completedTasks;//更新线程池的，线程执行完成数量
        workers.remove(w);//工作线程容器移除该工作线程
    } finally {
        mainLock.unlock();//解锁
    }
    //尝试结束线程池
    tryTerminate();
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {//如果当前线程池的运行状态是RUNNING\SHUTDOWN
        if (!completedAbruptly) {//如果该工作线程为正常结束
            /**
             * 判断当前需要的最少的核心线程数(如果允许核心线程超时，那么最小的核心线程数为0，否则为corePoolSize)
             */
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            //如果允许核心线程超时，而且等待队列不为空，那么工作线程的最小值为1，否则为0。
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            //当前工作线程数，是否满足最先的核心线程数
            if (workerCountOf(c) >= min)
                //如果满足那么直接return
                return; // replacement not needed
        }
        //如果是异常结束，或者当前线程数不满足最小的核心线程数，那么添加一个非核心线程
        //核心线程和非核心线程没有什么不同，只是在创建的时候判断逻辑不同
        addWorker(null, false);
    }
}

public void setCorePoolSize(int corePoolSize) {
    if (corePoolSize < 0)
        throw new IllegalArgumentException();
    //计算增量
    int delta = corePoolSize - this.corePoolSize;
    //覆盖原有的corePoolSize
    this.corePoolSize = corePoolSize;
    //如果当前的工作线程数量大于线程池的最大可运行核心线程数量，那么进行中断工作线程处理
    if (workerCountOf(ctl.get()) > corePoolSize)
        interruptIdleWorkers();
    else if (delta > 0) {//如果增量大于0
        // We don't really know how many new threads are "needed".
        // As a heuristic, prestart enough new workers (up to new
        // core size) to handle the current number of tasks in
        // queue, but stop if queue becomes empty while doing so.
        //等待队列非空，获取等待任务和增量的最小值
        int k = Math.min(delta, workQueue.size());
        //循环创建核心工作线程执行等待队列中的任务
        while (k-- > 0 && addWorker(null, true)) {
            if (workQueue.isEmpty())
                break;
        }
    }
}
private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();//加锁
    try {
        //遍历工作线程的集合
        for (Worker w : workers) {
            Thread t = w.thread;
            //如果当前线程没有被中断，而且能获取到锁，那么尝试进行中断，最后释放锁
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            //是否仅仅中断一个工作线程
            if (onlyOne)
                break;
        }
    } finally {//释放锁
        mainLock.unlock();
    }
}

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();// 检查权限
        advanceRunState(SHUTDOWN);// 设置线程池状态
        interruptIdleWorkers();// 中断空闲线程
        // 钩子函数，主要用于清理一些资源
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();// 检查状态
        advanceRunState(STOP);// 将线程池状态变为 STOP
        interruptWorkers();// 中断所有线程，包括工作线程以及空闲线程
        tasks = drainQueue();// 丢弃工作队列中存量任务
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}
private void interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)
            //如果工作线程已经开始，那么调用interrupt进行中断
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}
private List<Runnable> drainQueue() {
    BlockingQueue<Runnable> q = workQueue;
    ArrayList<Runnable> taskList = new ArrayList<Runnable>();
    //从此队列中删除所有可用的元素，并将它们添加到给定的集合中。
    q.drainTo(taskList);
    //如果队列是DelayQueue或其他类型的队列，而poll或drainTo可能无法删除某些元素，则会将它们逐个删除。
    if (!q.isEmpty()) {
        for (Runnable r : q.toArray(new Runnable[0])) {
            if (q.remove(r))
                taskList.add(r);
        }
    }
    return taskList;
}

// 关闭线程池的钩子函数
private static void shutdown(ExecutorService executorService) {
    // 第一步：使新任务无法提交
    executorService.shutdown();
    try {
        // 第二步：等待未完成任务结束
        if(!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
             // 第三步：取消当前执行的任务
            executorService.shutdownNow();
            // 第四步：等待任务取消的响应
            if(!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                System.err.println("Thread pool did not terminate");
            }
        }
    } catch(InterruptedException ie) {
        // 第五步：出现异常后，重新取消当前执行的任务
        executorService.shutdownNow();
        Thread.currentThread().interrupt(); // 设置本线程中断状态
    }
}