深入理解Java编程线程池的实现原理

public class ThreadPoolExecutor extends AbstractExecutorService {
  .....
  public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
      BlockingQueue<Runnable> workQueue);
 
  public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
      BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
 
  public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
      BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
 
  public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
    BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
  ...
}

TimeUnit.DAYS;        //天
TimeUnit.HOURS;       //小时
TimeUnit.MINUTES;      //分钟
TimeUnit.SECONDS;      //秒
TimeUnit.MILLISECONDS;   //毫秒
TimeUnit.MICROSECONDS;   //微妙
TimeUnit.NANOSECONDS;    //纳秒

ArrayBlockingQueue;
LinkedBlockingQueue;
SynchronousQueue;

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。 
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

public abstract class AbstractExecutorService implements ExecutorService {
 
  protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { };
  protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { };
  public Future<?> submit(Runnable task) {};
  public <T> Future<T> submit(Runnable task, T result) { };
  public <T> Future<T> submit(Callable<T> task) { };
  private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
              boolean timed, long nanos)
    throws InterruptedException, ExecutionException, TimeoutException {
  };
  public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
    throws InterruptedException, ExecutionException {
  };
  public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
              long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
  };
  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
    throws InterruptedException {
  };
  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                     long timeout, TimeUnit unit)
    throws InterruptedException {
  };
}

public interface ExecutorService extends Executor {
 
  void shutdown();
  boolean isShutdown();
  boolean isTerminated();
  boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException;
  <T> Future<T> submit(Callable<T> task);
  <T> Future<T> submit(Runnable task, T result);
  Future<?> submit(Runnable task);
  <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
    throws InterruptedException;
  <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                 long timeout, TimeUnit unit)
    throws InterruptedException;
 
  <T> T invokeAny(Collection<? extends Callable<T>> tasks)
    throws InterruptedException, ExecutionException;
  <T> T invokeAny(Collection<? extends Callable<T>> tasks,
          long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException;
}

public interface Executor {
  void execute(Runnable command);
}

execute()
submit()
shutdown()
shutdownNow()

volatile int runState;
static final int RUNNING  = 0;
static final int SHUTDOWN  = 1;
static final int STOP    = 2;
static final int TERMINATED = 3;

private final BlockingQueue<Runnable> workQueue;       //任务缓存队列，用来存放等待执行的任务
private final ReentrantLock mainLock = new ReentrantLock();  //线程池的主要状态锁，对线程池状态（比如线程池大小
                               //、runState等）的改变都要使用这个锁
private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集
 
private volatile long keepAliveTime;  //线程存货时间  
private volatile boolean allowCoreThreadTimeOut;  //是否允许为核心线程设置存活时间
private volatile int  corePoolSize;   //核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
private volatile int  maximumPoolSize;  //线程池最大能容忍的线程数
 
private volatile int  poolSize;    //线程池中当前的线程数
 
private volatile RejectedExecutionHandler handler; //任务拒绝策略
 
private volatile ThreadFactory threadFactory;  //线程工厂，用来创建线程
 
private int largestPoolSize;  //用来记录线程池中曾经出现过的最大线程数
 
private long completedTaskCount;  //用来记录已经执行完毕的任务个数

public void execute(Runnable command) {
  if (command == null)
    throw new NullPointerException();
  if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
    if (runState == RUNNING && workQueue.offer(command)) {
      if (runState != RUNNING || poolSize == 0)
        ensureQueuedTaskHandled(command);
    }
    else if (!addIfUnderMaximumPoolSize(command))
      reject(command); // is shutdown or saturated
  }
}

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
  Thread t = null;
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    if (poolSize < corePoolSize && runState == RUNNING)
      t = addThread(firstTask);    //创建线程去执行firstTask任务  
    } finally {
    mainLock.unlock();
  }
  if (t == null)
    return false;
  t.start();
  return true;
}

private Thread addThread(Runnable firstTask) {
  Worker w = new Worker(firstTask);
  Thread t = threadFactory.newThread(w); //创建一个线程，执行任务  
  if (t != null) {
    w.thread = t;      //将创建的线程的引用赋值为w的成员变量    
    workers.add(w);
    int nt = ++poolSize;   //当前线程数加1    
    if (nt > largestPoolSize)
      largestPoolSize = nt;
  }
  return t;
}

private final class Worker implements Runnable {
  private final ReentrantLock runLock = new ReentrantLock();
  private Runnable firstTask;
  volatile long completedTasks;
  Thread thread;
  Worker(Runnable firstTask) {
    this.firstTask = firstTask;
  }
  boolean isActive() {
    return runLock.isLocked();
  }
  void interruptIfIdle() {
    final ReentrantLock runLock = this.runLock;
    if (runLock.tryLock()) {
      try {
    if (thread != Thread.currentThread())
    thread.interrupt();
      } finally {
        runLock.unlock();
      }
    }
  }
  void interruptNow() {
    thread.interrupt();
  }
 
  private void runTask(Runnable task) {
    final ReentrantLock runLock = this.runLock;
    runLock.lock();
    try {
      if (runState < STOP &&
        Thread.interrupted() &&
        runState >= STOP)
      boolean ran = false;
      beforeExecute(thread, task);  //beforeExecute方法是ThreadPoolExecutor类的一个方法，没有具体实现，用户可以根据
      //自己需要重载这个方法和后面的afterExecute方法来进行一些统计信息，比如某个任务的执行时间等      
      try {
        task.run();
        ran = true;
        afterExecute(task, null);
        ++completedTasks;
      } catch (RuntimeException ex) {
        if (!ran)
          afterExecute(task, ex);
        throw ex;
      }
    } finally {
      runLock.unlock();
    }
  }
 
  public void run() {
    try {
      Runnable task = firstTask;
      firstTask = null;
      while (task != null || (task = getTask()) != null) {
        runTask(task);
        task = null;
      }
    } finally {
      workerDone(this);  //当任务队列中没有任务时，进行清理工作    
    }
  }
}

public void run() {
  try {
    Runnable task = firstTask;
    firstTask = null;
    while (task != null || (task = getTask()) != null) {
      runTask(task);
      task = null;
    }
  } finally {
    workerDone(this);
  }
}

Runnable getTask() {
  for (;;) {
    try {
      int state = runState;
      if (state > SHUTDOWN)
        return null;
      Runnable r;
      if (state == SHUTDOWN) // Help drain queue
        r = workQueue.poll();
      else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //如果线程数大于核心池大小或者允许为核心池线程设置空闲时间，
        //则通过poll取任务，若等待一定的时间取不到任务，则返回null
        r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
      else
        r = workQueue.take();
      if (r != null)
        return r;
      if (workerCanExit()) {  //如果没取到任务，即r为null，则判断当前的worker是否可以退出
        if (runState >= SHUTDOWN) // Wake up others
          interruptIdleWorkers();  //中断处于空闲状态的worker
        return null;
      }
      // Else retry
    } catch (InterruptedException ie) {
      // On interruption, re-check runState
    }
  }

private boolean workerCanExit() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  boolean canExit;
  //如果runState大于等于STOP，或者任务缓存队列为空了
  //或者 允许为核心池线程设置空闲存活时间并且线程池中的线程数目大于1
  try {
    canExit = runState >= STOP ||
      workQueue.isEmpty() ||
      (allowCoreThreadTimeOut &&
       poolSize > Math.max(1, corePoolSize));
  } finally {
    mainLock.unlock();
  }
  return canExit;
}

void interruptIdleWorkers() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    for (Worker w : workers) //实际上调用的是worker的interruptIfIdle()方法
      w.interruptIfIdle();
  } finally {
    mainLock.unlock();
  }
}

void interruptIfIdle() {
  final ReentrantLock runLock = this.runLock;
  if (runLock.tryLock()) {  //注意这里，是调用tryLock()来获取锁的，因为如果当前worker正在执行任务，锁已经被获取了，是无法获取到锁的
                //如果成功获取了锁，说明当前worker处于空闲状态
    try {
  if (thread != Thread.currentThread()) 
  thread.interrupt();
    } finally {
      runLock.unlock();
    }
  }
}

private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
  Thread t = null;
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    if (poolSize < maximumPoolSize && runState == RUNNING)
      t = addThread(firstTask);
  } finally {
    mainLock.unlock();
  }
  if (t == null)
    return false;
  t.start();
  return true;
}

public boolean prestartCoreThread() {
  return addIfUnderCorePoolSize(null); //注意传进去的参数是null
}
 
public int prestartAllCoreThreads() {
  int n = 0;
  while (addIfUnderCorePoolSize(null))//注意传进去的参数是null
    ++n;
  return n;
}

r = workQueue.take();

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

public class Test {
   public static void main(String[] args) {  
     ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
         new ArrayBlockingQueue<Runnable>(5));
 
     for(int i=0;i<15;i++){
       MyTask myTask = new MyTask(i);
       executor.execute(myTask);
       System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+
       executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());
     }
     executor.shutdown();
   }
}
 
class MyTask implements Runnable {
  private int taskNum;
 
  public MyTask(int num) {
    this.taskNum = num;
  }
 
  @Override
  public void run() {
    System.out.println("正在执行task "+taskNum);
    try {
      Thread.currentThread().sleep(4000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("task "+taskNum+"执行完毕");
  }
}

正在执行task 0
线程池中线程数目：1，队列中等待执行的任务数目：0，已执行玩别的任务数目：0
线程池中线程数目：2，队列中等待执行的任务数目：0，已执行玩别的任务数目：0
正在执行task 1
线程池中线程数目：3，队列中等待执行的任务数目：0，已执行玩别的任务数目：0
正在执行task 2
线程池中线程数目：4，队列中等待执行的任务数目：0，已执行玩别的任务数目：0
正在执行task 3
线程池中线程数目：5，队列中等待执行的任务数目：0，已执行玩别的任务数目：0
正在执行task 4
线程池中线程数目：5，队列中等待执行的任务数目：1，已执行玩别的任务数目：0
线程池中线程数目：5，队列中等待执行的任务数目：2，已执行玩别的任务数目：0
线程池中线程数目：5，队列中等待执行的任务数目：3，已执行玩别的任务数目：0
线程池中线程数目：5，队列中等待执行的任务数目：4，已执行玩别的任务数目：0
线程池中线程数目：5，队列中等待执行的任务数目：5，已执行玩别的任务数目：0
线程池中线程数目：6，队列中等待执行的任务数目：5，已执行玩别的任务数目：0
正在执行task 10
线程池中线程数目：7，队列中等待执行的任务数目：5，已执行玩别的任务数目：0
正在执行task 11
线程池中线程数目：8，队列中等待执行的任务数目：5，已执行玩别的任务数目：0
正在执行task 12
线程池中线程数目：9，队列中等待执行的任务数目：5，已执行玩别的任务数目：0
正在执行task 13
线程池中线程数目：10，队列中等待执行的任务数目：5，已执行玩别的任务数目：0
正在执行task 14
task 3执行完毕
task 0执行完毕
task 2执行完毕
task 1执行完毕
正在执行task 8
正在执行task 7
正在执行task 6
正在执行task 5
task 4执行完毕
task 10执行完毕
task 11执行完毕
task 13执行完毕
task 12执行完毕
正在执行task 9
task 14执行完毕
task 8执行完毕
task 5执行完毕
task 7执行完毕
task 6执行完毕
task 9执行完毕

Executors.newCachedThreadPool();    //创建一个缓冲池，缓冲池容量大小为Integer.MAX_VALUE
Executors.newSingleThreadExecutor();  //创建容量为1的缓冲池
Executors.newFixedThreadPool(int);  //创建固定容量大小的缓冲池

public static ExecutorService newFixedThreadPool(int nThreads) {
  return new ThreadPoolExecutor(nThreads, nThreads,
                 0L, TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
  return new FinalizableDelegatedExecutorService
    (new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                 60L, TimeUnit.SECONDS,
                 new SynchronousQueue<Runnable>());
}