深入理解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>());
}