线程池

在java中，如果每个请求到达就创建一个新线程，创建和销毁线程花费的时间和消耗的系统资源都相当大，甚至可能要比在处理实际的用户请求的时间和资源要多的多。

​ 如果在一个JVM里面创建太多的线程，可能会使系统由于过度消耗内存或”切换过度“而导致系统资源不足。

​ 为了解决这个问题，有了线程池的概念。线程池的核心逻辑是提前创建好若干个线程放在一个容器中。如果有任务需要处理，则将任务直接分配给线程池中的线程来执行就行，任务处理以后这个线程不会被销毁，而是等待后续分配任务。同时通过线程池来重复管理线程还可以避免创建大量线程增加开销。

线程池的优势：

合理的使用线程池，可以带来一些好处

1.降低创建线程和销毁线程的性能开销

2.提高响应速度，当有新任务需要执行是不需要等待线程创建就可以立马执行

3.合理的设置线程池大小可以避免因为线程数超过硬件资源瓶颈带来的问题

Java中提供的线程池 API

线程池的使用

JDK为我们提供了几种不同的线程池实现。Java是如何创建线程池呢？

举例如下：创建三个固定线程数的线程池

public class Test implements Runnable{

    public static ExecutorService executor = Executors.newFixedThreadPool(3);
    public static void main(String[] args) {

        for (int i = 0; i < 1000; i++) {
            executor.execute(new Test());
        }
        executor.shutdown();
    }

    @Override
    public void run() {
        try {
            Thread.sleep(0);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName());
    }
}

Java中提供的线程池API

在Executors里面提供了几个线程池的工厂方法，只需要直接使用Executors的工厂方法，就可以使用线程池：

newFixedThreadPool:该方法返回一个固定数量的线程池，线程数不变，当有一个任务提交时，若线程池中空闲，则立即执行，若没有，则会被在暂缓在一个任务队列中，等待有空闲的线程去执行。

newSingleThreadExecutor:创建一个线程的线程池，若空闲则执行，若没有空闲线程则暂缓在任务队列中。

newCachedThreadPool:返回一个可根据实际情况调整线程个数的线程池，不限制最大线程数量，若用空闲的线程则执行任务，若没有任务则不创建线程。并且每个空闲线程会在60秒后自动回收。

newScheduledThreadPool:创建一个可以指定线程的数量的线程池，但是这个线程池还带有延迟和周期性执行任务的功能，类似定时器。

ThreadPoolExecutor

上面提到的四种线程池的构建，都是基于ThreadPoolExecutor来构建。

面试题：简单说一下你知道的线程池和ThreadPoolThread有哪些构造参数？

ThreadPoolExecutor有多个重载的构造方法，我们可以基于他最完整的构造方法来分析。

先解释一下每个参数的作用：

  public ThreadPoolExecutor(int corePoolSize, //核心线程数
                              int maximumPoolSize, //最大线程数
                              long keepAliveTime, //超出时间，超出核心线程数量以外的线程空余存活时间
                              TimeUnit unit, //存活时间单位
                              BlockingQueue<Runnable> workQueue, //保存执行任务的队列
                              ThreadFactory threadFactory, //创建新线程使用的工厂
                              RejectedExecutionHandler handler)  //当任务无法执行的时候处理方式

线程初始化以后做了什么事情？

线程池初始化时是没有创建线程的，线程池里的线程的初始化于其他线程一样，但是在完成任务以后，该线程不会自行销毁，而是以挂起的状态返回到线程池。直到应用程序再次向线程池发出请求时，线程池里挂起的线程就会再度激活执行任务。这样既节省了建立线程所造成的的性能损耗，也可以让多个任务反复重用同一个线程，从而在应用程序生存期内节约大量开销。

newFixedThreadPool

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

FixedThreadPool的核心线程数和最大线程数都是指定值，也就是说当前线程池中的线程数超过了核心线程数后，任务都会被放到阻塞队列中。另外keepAliveTime为0，也就是超出核心线程数量以外的线程空余存活时间

而这里选用的阻塞队列是LinkedBlockingQueue,使用的是默认容器Interger.MAX_VALUE,相当于没有使用上限

这里的线程池执行任务的流程如下：

1.线程数小于核心线程数，也就是设置的线程数时，新建线程执行任务

2.线程数等于和核心线程数后，将任务加入阻塞队列

3.由于线程容量非常大，可以一直添加

4.执行任务的线程反复去队列中去任务执行

用途：FixedThreadPool用于负载比较大的服务器，为了资源的合理利用，需要限制当前线程数量

newCachedThreadPool

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

CachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程；并且没有核心线程，非核心线程数无上限，但每个空闲的时间只有60秒，超过后就会被回收。

它的执行流程如下：

1.没有核心线程，直接向SynchronousQueue中提交任务

2.如果有空闲线程，就去取出任务执行；如果没有空闲线程，就新建一个

3.执行完任务的线程有60秒生存时间，如果在这个时间内可以接收到新任务，就可以继续活下去，否则就被回收

newSingleThreadExecutor

创建一个单线程化的线程池，他只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序（FIFO,LIFO,优先级）执行

线程池的实现原理分析

ThreadPoolExecutor是线程池的核心，提供了线程池的实现。

ScheduledThreadPoolExecutor继承了ThreadPoolExrcutor，并另外提供了一些调度方法以支持定时和周期任务。Executers是工具类，主要用来创建线程池对象。

我们把一个任务提交给线程池去处理的时候，线程池的处理过程是怎样的呢？

线程池原理分析：

源码分析

execute方法

public void execute(Runnable command) {
  if (command == null)
    throw new NullPointerException();
  int c = ctl.get();
  //1.当前池中线程比核心数少，新建一个线程执行任务
  if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
      return;
    c = ctl.get();
  }
  //2.核心池已满，但任务队列未满，添加到队列中
  if (isRunning(c) && workQueue.offer(command)) {
    int recheck = ctl.get();
    //任务成功添加到队列以后，再次检查是否需要添加新的线程，因为已存在的线程可能被销毁了
    if (! isRunning(recheck) && remove(command))
      //如果线程池处于非运行状态，并且把当前的任务从任务队列中移除成功，则拒绝该任务
      reject(command);
    //如果之前的线程已经被销毁完，新建一个线程
    else if (workerCountOf(recheck) == 0)
      addWorker(null, false);
  }
  //3.核心池已满，队列已满，试着创建一个新线程
  else if (!addWorker(command, false))
    //如果创建新线程失败了，说明线程池被关闭了或者线程池完全满了，拒绝任务
    reject(command);
}

ctl的作用

在线程池中，ctk贯穿在线程池的整个生命周期中

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

它是一个原子类，主要作用是用来保存线程数量和线程池的状态。代码分析，他这里用到了位运算。

一个int数值是32个bit位，这里采用高3位来保存运行状态，低29位来保存线程数量。

我们分析默认情况下。也就是ctlOf(RUNNING)运行状态，调用了ctlOf(int rs,int wc)方法；

private static int ctlOf(int rs, int wc) { return rs | wc; }

其中RUNNING = -1 << COUNT_BITS; -1左移29位，-1的二进制是32个1（11.。。。1111）

-1的二进制计算方法
原码是1000..001 高位1表示符号位。
然后对原码取反，高位不变得到1111...110
然后对反码进行+1，也就是补码操作，最后得到1111...1111

那么-1<<左移29位，也就是【111】表示；rs|wc。二进制的111|000。得到的结果仍然是111

那么同理可得其他的状态bit为表示

private static final int COUNT_BITS = Integer.SIZE - 3; //32 - 3 
private static final int CAPACITY   = (1 << COUNT_BITS) - 1; //将1的二进制向左移动29位，再减1表示最大线程容量

//运行状态保存在int值得高3位（所有数值左移29位）
private static final int RUNNING    = -1 << COUNT_BITS; //接受新任务，并执行队列中的任务
private static final int SHUTDOWN   =  0 << COUNT_BITS; //不接受新任务，但是执行队列中的任务
private static final int STOP       =  1 << COUNT_BITS; //不接受新任务，不执行任务中的任务，中断正在执行中的任务
private static final int TIDYING    =  2 << COUNT_BITS; //所有的任务都已结束，线程数量为0，处于该状态的线程池即将调用terminated()方法
private static final int TERMINATED =  3 << COUNT_BITS; //terminated()方法执行完成

状态转化

addWork

如果工作线程数小于核心线程数的话，会调用addWorker,顾名思义，其实就是要创建一个工作线程。

源码实现：

1）采用循环CAS操作来将线程数加1；

2）新建一个线程并启用。

private boolean addWorker(Runnable firstTask, boolean core) {
  //goto语句，避免死循环
  retry:
  for (;;) {
    int c = ctl.get();
    int rs = runStateOf(c);

    // Check if queue empty only if necessary.
    //如果线程处于非运行状态，并且rs不等于SHUTDOWN且firstTask不等于空并且workQueue为空，直接返回false(表示不可添加work状态)
    //1.线程池已经shutdown后，还要添加新的任务，拒绝
    //2.SHUTDOWNh状态不接受新任务，但仍然会执行已经加入队列的任务，所以当进入SHUTDOWN状态，而传进来的任务为空，并且任务队列不为空的时候，是允许添加新线程的，如果把这个状态取反，就表示不允许添加worker
    if (rs >= SHUTDOWN &&
        ! (rs == SHUTDOWN &&
           firstTask == null &&
           ! workQueue.isEmpty()))
      return false;

    //自旋
    for (;;) {
      //获得Worker工作线程数
      int wc = workerCountOf(c);
      //如果工作线程数大于默认容量大小或者大于核心/最大线程数大小，则直接返回false表示不能再添加worker.
      if (wc >= CAPACITY ||
          wc >= (core ? corePoolSize : maximumPoolSize))
        return false;
      //通过cas来增加工作线程数，如果增加成功，直接跳出循环，否则在检查线程池状态后，再次在内循环调用compareAndIncrementWorkerCount，直到添加成功
      if (compareAndIncrementWorkerCount(c))
        break retry;
      //再次获取ctl的值
      c = ctl.get();  // Re-read ctl
      //这里如果不相等，说明线程的状态发生了变化，继续重试
      if (runStateOf(c) != rs)
        continue retry;
      // else CAS failed due to workerCount change; retry inner loop
    }
  }

  //上面的代码主要是对worker数量做原子+1操作，下面的逻辑才是正式构建一个worker
  boolean workerStarted = false; //工作线程是否启动的标识
  boolean workerAdded = false; //工作线程是否已经添加成功的标识
  Worker w = null;
  try {
    //构建一个Worker,这个worker是什么呢？我们可以看到构造方法里面传入了一个Runnable对象
    w = new Worker(firstTask);
    //从worker对象中取出线程
    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());

        //只有当前线程池是正在运行状态，或是SHUTDOWN且firstTask为空，才能添加到workers集合中
        if (rs < SHUTDOWN ||
            (rs == SHUTDOWN && firstTask == null)) {
          //任务刚封装到work里面，还没start,你封装的线程就是alive，需要抛异常出去
          if (t.isAlive()) // precheck that t is startable
            
            throw new IllegalThreadStateException();
         //将新创建的Worker添加到workers集合中
          workers.add(w);
          //如果集合中的工作线程数大于最大线程数，这个最大线程数表示线程池曾经出现过的最大线程数
          int s = workers.size();
          if (s > largestPoolSize)
            //更新线程池出现过的最大线程数
            largestPoolSize = s;
          //表示工作线程创建成功了
          workerAdded = true;
        }
      } finally {
        mainLock.unlock();
      }
      //如果worker添加成功
      if (workerAdded) {
        //启动线程
        t.start();
        workerStarted = true;
      }
    }
  } finally {
    if (! workerStarted)
      //如果添加失败，就需要做一件事，就是递减实际工作线程（正如开始的时候增加的工作线程数）
      addWorkerFailed(w);
  }
  //返回结果
  return workerStarted;
}

Worker类

我们发现addWorker方法只是一个构造了Worker，并且把firstTask封装到worker中，它是做什么的呢？

1.每个worker,都是一条线程，同时里面包含了一个firstTask,即初始化时要被首先执行的任务；

2.最终执行任务的是runWorker()方法

Worker类继承了AQS，并实现了Runnable接口，注意其中的firstTask和thread属性；

firstTask用它来保存传入的任务；thread是在调用构造方法时通过ThreadFactory来创建的线程，是用来处理任务的线程。

在调用构造方法时，需要传入任务，这里通过getThreadFactory.newThread(this);来创建一个线程，newThread方法传入的参数是this,因为Worker本身继承了Runnable接口，也就是一个线程，所以一个Worker对象在启动的时候会调用Worker类中的run方法。

Worker继承了AQS，使用AQS来实现独占锁的功能。问什么使用ReentrantLock来实现呢？可以看到tryAcquire方法，它是不允许重入的，而ReentrantLock是允许重入的：

Lock方法一旦获取了独占锁，表示当前线程正在执行任务中；那么它会有以下几个作用：

1.如果正在执行任务，则不应该中断线程；

2.如果该线程现在不是独占锁的状态，也就是空闲的状态，说明他没有在处理任务，这时可以对该线程进行中断；

3.线程池在执行shutdown方法或tryTerminate方法时会调用interruptIdleWorkers方法来中断空闲的线程，interruptIdleWorkers方法会使用tryLock方法来判断线程池中的线程是否是空闲状态

4.之所以设置为不可重入，是因为我们不希望任务在调用像setCorePoolSize这样的线程池控制方法时重新获取锁，这样会中断正在运行的线程。

private final class Worker
  extends AbstractQueuedSynchronizer
  implements Runnable {
  
  private static final long serialVersionUID = 6138294804551838833L;

  /** Thread this worker is running in.  Null if factory fails. */
  //注意！！这才是真正执行task的线程，从构造函数可知是由ThreadFactory创建的
  final Thread thread;
  //这就是需要执行的task
  /** Initial task to run.  Possibly null. */
  Runnable firstTask;
  /** Per-thread task counter */
  //完成的任务数，用于线程池统计
  volatile long completedTasks;

  Worker(Runnable firstTask) {
    //初始状态-1，防止在调用runWork()，也就是真正执行task前中断thread.
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
  }

  /** Delegates main run loop to outer runWorker  */
  public void run() {
    runWorker(this);
  }

  // Lock methods
  //
  // The value 0 represents the unlocked state.
  // The value 1 represents the locked state.

  protected boolean isHeldExclusively() {
    return getState() != 0;
  }

  protected boolean tryAcquire(int unused) {
    if (compareAndSetState(0, 1)) {
      setExclusiveOwnerThread(Thread.currentThread());
      return true;
    }
    return false;
  }

  protected boolean tryRelease(int unused) {
    setExclusiveOwnerThread(null);
    setState(0);
    return true;
  }

  public void lock()        { acquire(1); }
  public boolean tryLock()  { return tryAcquire(1); }
  public void unlock()      { release(1); }
  public boolean isLocked() { return isHeldExclusively(); }

  void interruptIfStarted() {
    Thread t;
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
      try {
        t.interrupt();
      } catch (SecurityException ignore) {
      }
    }
  }
}

addWorkerFailed

addWorker方法中，如果添加Worker并且启动线程失败，则会做失败后的处理。

这个方法主要做两件事：

1.如果worker已经构建好了，则从workers集合中移除这个worker

2.原子递减核心线程数（因为在addWorker方法先做了原子增加）

3.尝试结束线程池

private void addWorkerFailed(Worker w) {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    if (w != null)
      workers.remove(w);
    decrementWorkerCount();
    tryTerminate();
  } finally {
    mainLock.unlock();
  }
}

runWorker方法

前面了解了ThreadPoolExecutor的核心方法addWorker,主要作用是增加工作线程，而Worker简单理解其实就是一个线程，里面实现了run方法，这块是线程中执行任务的真正处理逻辑，也就是runWorker方法，这个方法主要做几件事：

1.如果task不为空，则开始执行task

2.如果task为空，则通过getTask()再去取任务，并赋值给task，如果取到的Runnable不为空，则执行该任务

3.执行完毕后，通过while循环继续getTask()取任务

4.如果getTask()取到的任务依然是空，那么整个runWorker()方法执行完毕

final void runWorker(Worker w) {
  Thread wt = Thread.currentThread();
  Runnable task = w.firstTask;
  w.firstTask = null;
  //unlock，表示当前worker线程允许中断，因为new Worker默认的state =-1,此处是调用Worker类的tryRelease()方法，将state置位0；而interruptIfStarted()中只有state>=0才允许调用中断
  w.unlock(); // allow interrupts
  boolean completedAbruptly = true;
  try {
    //注意这个while循环，在这里实现了【线程复用】 如果task为空，则通过getTask来获取任务
    while (task != null || (task = getTask()) != null) {
      //上锁，不是为了防止并发执行任务，为了在shutdown()时不终止正在运行的worker
      w.lock();
      //线程池为stop状态时不接受新任务，不执行已经加入任务队列的任务，还中断正在执行的任务
      //所以对于stop状态以上是要中断线程的
      //（Thread.interrupted() && runStateAtLeast()ctl.get(), STOP）确保线程中断标志位为true且是stop状态以上，接着清除了中断标志
      //！wt.isInterrupted()则再一次检查保证线程需要设置中断标志位      
      // 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
      if ((runStateAtLeast(ctl.get(), STOP) ||
           (Thread.interrupted() &&
            runStateAtLeast(ctl.get(), STOP))) &&
          !wt.isInterrupted())
        wt.interrupt();
      try {
        //这里默认是没有实现的，在一些特定的场景中我们可以自己继承ThreadPoolExecutor自己重写
        beforeExecute(wt, task);
        Throwable thrown = null;
        try {
          //执行任务中的run方法
          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，需要再通过getTask()取+记录该worker完成任务数量 + 解锁）
        task = null;
        w.completedTasks++;
        w.unlock();
      }
    }
    completedAbruptly = false;
  } finally {
    //1.将入参worker从数组workers里删掉；
    //2.根据布尔值allowCoreThreadTimeOut来决定是否补充新的Worker进数组workers
    processWorkerExit(w, completedAbruptly);
  }
}

getTask

worker线程会从阻塞队列中获取需要执行的任务，这个方法不是简单的take任务。

怎么判断线程有多久没有活动了，是不是以为线程池会启动一个监控线程，专门监控哪个线程不干活？

答案是：在线程从工作队列poll任务时，加上了超时限制，如果线程在keepAliveTime的时间内poll不到任务，那么我们就认为这条线程没事做，就可以销毁了。

private Runnable getTask() {
  boolean timedOut = false; // Did the last poll() time out?

  //自旋
  for (;;) {
    int c = ctl.get();
    int rs = runStateOf(c);
    //对线程池状态的判断，两种情况会workerCount-1,并且返回null
    //1.线程池状态为shutdown，并且workQueue为空（反映了shutdown状态的线程池还是要执行workQueue中剩余的任务的）
    //2.线程池状态为stop（shutdownNow()会导致变成STOP）(此时不用考虑workQueue的情况)

    // Check if queue empty only if necessary.
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
      decrementWorkerCount();
      //返回null，则当前worker线程会退出
      return null;
    }

    int wc = workerCountOf(c);

    // Are workers subject to culling?
    //timed变量用于判断是否需要进行超时控制。
    //allowCoreThreadTimeOut默认是false，也就是核心线程不允许进行超时；
    // wc > corePoolSize,表示当前线程池中的线程数量大于核心线程数量；
    //对于超过核心线程数量的这些线程，需要进行超时控制
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

    //1.线程数量超过maxmumPoolSize可能是线程池在运行时被调用了setMaxiumumPoolSize()被改变了大小，否则已经addWorker()成功不会超过maximumPoolSize
    //2.time && timeout如果为true，表示当前操作需要进行超时控制，并且上次从阻塞队列中获取任务发生了超时。其实就是体现了空闲线程的存活时间
    if ((wc > maximumPoolSize || (timed && timedOut))
        && (wc > 1 || workQueue.isEmpty())) {
      if (compareAndDecrementWorkerCount(c))
        return null;
      continue;
    }

    try {
      //根据timed来判断，如果为true，则通过阻塞队列poll方法进行超时控制，如果在keepAliveTime时间内没有获取到任务，则返回null.否则通过take方式阻塞式获取队列中的任务
      Runnable r = timed ?
        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
      workQueue.take();
      //如果拿到的任务不为空，则直接返回给worker进行处理
      if (r != null)
        return r;
      //如果r==null,说明已经超时了，设置timedOut=true，在下次自旋的时候进行回收
      timedOut = true;
    } catch (InterruptedException retry) {
      //如果获取任务时当前线程发生了中断，则设置timedOut为false并返回循环重试
      timedOut = false;
    }
  }
}

这里重要的地方是第二个if判断，目的是控制线程池的有效线程数量。由上文中的分析可以知道，在执行execute方法时，如果当前线程池的线程数量超过了corePoolSize且小于maximumPoolSize，并且workQueue已满时，则可以增加工作线程，但这时如果超时没有获取到任务，也就是timeOut为true的情况，说明workQueue已经空了，也就说明了当前线程不需要那么多线程来执行任务了，可以把多于corePoolSize数量的线程销毁掉，保持线程数量在corePoolSize即可。

什么时候会销毁呢？当然是runWorker方法执行完之后，也就是Worker中的run方法执行完，由JVM自动回收。

getTask方法返回null时，在runWorker方法中 会跳出while循环，然后会执行processWorkerExit方法。

processWorkerExit

runWorker的while循环执行完毕以后，在finally中会调用processWorkerExit，来销毁工作线程。

到目前为止，我们已经从execute方法中输入了worker线程的创建到执行以及最后到销毁的全部过程。那么我们继续回到execute方法。

execute后续逻辑分析

如果核心线程数已满，说明这个时候不能再创建核心线程了，于是走第二个判断，第二个段逻辑相对简单，如果线程池处于运行状态并且任务队列没有满，则将任务添加到队列中，第三个判断，核心线程数满了，队列也满了，那么这个时候创建新的线程池也就是（非核心线程）

如果非核心线程数也达到了最大线程数大小，直接拒绝任务。

//核心池已满，但任务队列未满，添加到队列中
if (isRunning(c) && workQueue.offer(command)) {
  int recheck = ctl.get();
  //任务成功添加到队列以后，再次检查是否需要添加新的线程，因为已经存在的线程可能被销毁了
  if (! isRunning(recheck) && remove(command))
    //如果当线程池处于非运行状态，并且把当前的任务从任务队列中移除成功，则拒绝该任务
    reject(command);
  //如果之前的线程已经被销毁完了，创建一个线程
  else if (workerCountOf(recheck) == 0)
    addWorker(null, false);
}
//核心线程池已满，队列已满，试着创建一个新线程
else if (!addWorker(command, false))
  //如果创建新线程失败了，说明线程池被关闭了或者线程池完全满了，拒绝任务
  reject(command);

拒绝策略

1.AbortPolicy：直接抛出异常，默认策略；

2.CallerRunsPolicy:用调用者所在的线程来执行任务

3.DiscardOldestPolicy:丢弃阻塞队列中靠前的任务，并执行当前任务；

4.DiscardPolicy:直接丢弃任务；

当然也可以根据应用场景实现RejectedExecutionHandler接口，自定义饱和策略，如记录日志或持久化存储不能处理的任务。

线程池的注意事项

QA: 线程池的构建不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式。

原理分析:使用Executors使得用户不用关系线程池的参数配置，意味着大家对于线程池的运行规则也会慢慢的忽略。这会导致一个问题，比如我们使用newFixedThreadPool或者singleThreadPool允许的队列长度为Integer.MAX_VALUE,如果使用不当会导致大量请求堆积到队列中导致OOM的风险。而newCachedThreadPool，允许创建线程数量为Integer.MAX_VALUE，也可能会导致大量线程的创建出现CPU使用过高或者OOM的问题。

而如果我们通过ThreadPoolExecutor来构造线程池的话，我们势必要了解线程池构造中每个参数的具体含义，使的开发者在配置参数的时候能够更加谨慎。

合理配置线程池的大小

分析：

1.需要分析线程池执行的任务的特性：CPU密集型还是IO密集型

2.每个任务执行的平均时长大概是多少，这个任务的执行时长可能还跟任务处理逻辑是否涉及到网络传输以及底层系统资源依赖有关系

如果是CPU密集型，主要是执行计算任务，响应时间很快，cpu一直在运行，这种任务cpu的利用率很高，那么线程数的配置应该根据CPU核心数来决定，CPU核心数=最大同时执行线程数，加入CPU核心数为4，那么服务器最多能同时执行4个线程。过多的线程会导致上下文切换反而使得效率降低。那线程池的最大线程数可以配置为cpu核心数+1

如果是IO密集型，主要进行IO操作，执行IO操作的时间较长，这是cpu处于空闲状态，导致cpu的利用率不高，这种情况下可以增加线程池的大小。这种情况下可以结合线程的等待时长来做判断，等待时间越高，那么线程数也相对越多。一般可以配置cpu核心数的2倍。

一个公式：线程池设定最佳线程数目= ((线程池设定的线程的等待时间+线程CPU时间) / 线程CPU) * CPU数目

这个公式的线程cpu时间是预估的程序单个线程在CPU上运行的时间（通常使用loadRunner测试大量运行次数求出平均值）

默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个方法办到：

prestartCoreThread():初始化一个核心线程；

prestartAllCoreThread():初始化所有核心线程

ThreadPoolExecutor tpe = （ThreadPoolExecutor）service;
tpe.prestartAllCoreThread();

线程池的关闭

ThreadPoolExecutor提供了两个方法，用于线程池的关闭，分别是shutdown()和shutdownNow(),其中：shutdown():不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才会终止，但再也不会接受新的任务

shutdownNow():立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务。

线程池容量的动态调整

ThreadPoolExecutor提供了动态调整线程池容量大小的方法：setCorePoolSize()和setMaximumPoolSize(),

setCorePoolSize:设置核心池大小

setMaximumPoolSize:设置线程池最大能创建的线程数目大小

任务缓存队列及排队策略

任务缓存队列，即workQueue,他用来存放等待执行的任务。workQueue的类型类BlockingQueue,通常可以取下面三种类型：

ArrayBlockingQueue:基于数组的先进先出队列，此队列创建时必须指定大小；

LinkedBlockingQueue:基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE;

SynchronousQueue:这个队列比较特殊，他不会保存提交的任务，而是将直接创建一个线程来执行新来的任务。

线程池的监控

如果在项目中大规模的使用了线程池，那么必须有一套监控体系，来知道当前线程池的状态，当出现问题的时候可以快速定位到问题。而线程池提供了相应的扩展方法，我们通过重写线程池的beforeExecute、afterExecutor和shutdown等方式就可以实现对线程的监控。

示例：

public class ThreadPoolDemo extends ThreadPoolExecutor {

  //保持任务开始执行的时间，当任务结束时，用任务结束时间减去开始时间计算任务执行时间
  private ConcurrentHashMap<String, Date> startTimes;

  public ThreadPoolDemo(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    this.startTimes = new ConcurrentHashMap<>();
  }


  @Override
  public void shutdown() {
    System.out.println("已经执行的任务数：" + this.getCompletedTaskCount() + "," + "当前活动线程数" + this.getActiveCount()
                       + ",当前排队线程数" + this.getQueue().size());

    System.out.println();
    super.shutdown();
  }

  /**
     * 任务开始之前记录任务开始时间
     *
     * @param t
     * @param r
     */
  @Override
  protected void beforeExecute(Thread t, Runnable r) {
    startTimes.put(String.valueOf(r.hashCode()), new Date());
    super.beforeExecute(t, r);
  }


  @Override
  protected void afterExecute(Runnable r, Throwable t) {
    Date startDate = startTimes.remove(String.valueOf(r.hashCode()));
    Date finishDate = new Date();
    long diff = finishDate.getTime() - startDate.getTime();
    System.out.print("任务耗时:" + diff + "
");
    System.out.print("初始线程数:" + this.getPoolSize() + "
");
    System.out.print("核心线程数:" + this.getCorePoolSize() + "
");
    System.out.print("正在执行的任务数量:" + this.getActiveCount() + "
");
    System.out.print("已经执行的任务数:" + this.getCompletedTaskCount() + "
");
    System.out.print("任务总数:" + this.getTaskCount() + "
");
    System.out.print("最大允许的线程数:" + this.getMaximumPoolSize() + "
");
    System.out.print("线程空闲时间:" + this.getKeepAliveTime(TimeUnit.MILLISECONDS) + "
");
    System.out.println();
    super.afterExecute(r, t);
  }

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

测试脚本

public class ThreadPoolDemoTest implements Runnable{

    private static ExecutorService es = ThreadPoolDemo.newCachedThreadPool();

    @Override
    public void run() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        for (int i = 0; i < 100; i++) {
            es.execute(new Test());
        }
        es.shutdown();
    }
}

Callable/Future使用及原理分析

线程的执行任务有两种方法，一种是submit，一种是execute;这两个方法是有区别的。

execute和submit区别

1.execute只可以接受一个Runnable的参数

2.execute如果出现异常会抛出

3.execute没有返回值

1.submit可以接收Runable和Callable这两种类型的参数，

2.对于submit方法，如果传入一个Callable,可以得到一个Future的返回值

3.submit方法调用不会抛出异常，除非调用Future.get

下面分析Callable/Future:

Callable/Future案例演示

Callable/Future和Thread之类的线程构建最大的区别在于，能够很方便的获取线程执行完以后的结果。

public class CallableDemo implements Callable<String> {

    @Override
    public String call() throws Exception {
        //Thread.sleep(10000);
        return "hello World";
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CallableDemo callableDemo = new CallableDemo();
        FutureTask futureTask = new FutureTask(callableDemo);
        new Thread(futureTask).start();
        System.out.println(futureTask.get());
    }
}

问什么要使用回调呢？那是因为结果值是由另一线程计算的，当前线程是不知道结果值什么时候计算完成，所以它传递一个回调接口给计算线程，当计算完成时，调用这个回调接口，回传结果值。

这个在很多地方用到，比如Dubbo的异步调用，比如消息中间件的异步通信等等。。。

利用FutureTask、Callable、Thread对耗时任务（如查询数据库）做预处理，在需要计算结果之前就启动计算。

Callable/Future原理分析

Callable是一个函数式接口，里面就只有一个call方法。子类可以重写这个方法，并且这个方法会有一个返回值

@FunctionalInterface
public interface Callable<V> {
    V call() throws Exception;
}

FutureTask

FutureTask的类关系图如下，它实现RunnableFuture接口，那么这个RunnableFuture接口的作用？

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

RunnableFuture是一个接口，他继承了Runnable和Future这两个接口，Runnable很熟悉了，那么Future是什么呢？

Future表示一个任务的生命周期，并提供了相应的方法来判断是否已经完成或取消，以及获取任务的结果和取消任务等。

*/
public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    //当前的Future是否被取消，返回true表示已取消
    boolean isCancelled();

    //当前Future是否已结束。包括运行完成、抛出异常以及取消，都表示当前Future已结束
    boolean isDone();

    //获取Future的结果值。如果当前Future还没有结束，那么当前线程就等待，
    //直到Future运行结束，就会唤醒等待结果值的线程的。
    V get() throws InterruptedException, ExecutionException;

    //获取Future的结果值。与get()相比较多了允许设置超时时间
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

到这里我们初步了解了FutureTask是Runnable和Future的结合，如果我们把Runnable比作是生产者，Future比作是消费者，那么FutureTask是被这两者共享的，生产者运行run方法计算结果，消费者通过get方法获取结果。

作为生产者消费者模式，有一个很重要的机制，就是如果生产者数据还没准备的时候，消费者会被阻塞。当生产者数据准备好了以后会唤醒消费者继续执行。

state的含义

表示FutureTask当前的状态，分为7中状态

private volatile int state;
//New新建状态，表示这个FutureTask还没有开始运行
private static final int NEW          = 0;
//COMPLETING完成状态，表示FutureTask任务已经计算完毕了
//但是还有一些后续操作，例如唤醒等待线程操作，还没有完成。
private static final int COMPLETING   = 1;
//FutureTask任务完结，正常完成，没有发生异常
private static final int NORMAL       = 2;
//FutureTask任务完结，因为发生异常
private static final int EXCEPTIONAL  = 3;
//FutureTask任务完结，因为取消任务
private static final int CANCELLED    = 4;
//FutureTask任务完结，也是取消任务，不过发起了中断运行任务线程的中断请求
private static final int INTERRUPTING = 5;
//FutureTask任务完结，也是取消任务，已经完成了中断运行任务线程的中断请求
private static final int INTERRUPTED  = 6;

run方法

public void run() {
  //如果状态state不是NEW，或者设置runner值失败
  //表示有别的线程在此之前调用run方法，并成功设置了runner值
  //保证了只有一个线程可以运行try代码块中的代码
  if (state != NEW ||
      !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                   null, Thread.currentThread()))
    return;
  try {
    Callable<V> c = callable;
    //只有c不为null且状态state为New的情况
    if (c != null && state == NEW) {
      V result;
      boolean ran;
      try {
        //调用callable的call方法，并获得返回结果
        result = c.call();//运行成功
        ran = true;
      } catch (Throwable ex) {
        result = null;
        ran = false;
        //设置异常结果
        setException(ex);
      }
      if (ran)
        //设置结果
        set(result);
    }
  } finally {
    // runner must be non-null until state is settled to
    // prevent concurrent calls to run()
    runner = null;
    // state must be re-read after nulling runner to prevent
    // leaked interrupts
    int s = state;
    if (s >= INTERRUPTING)
      handlePossibleCancellationInterrupt(s);
  }
}

其实run方法作用非常简单，就是调用callable的call方法返回结果值result，根据是否发生异常，调用set(result)或setException(ex)方法表示FutureTask任务完结。

不过因为FutureTask任务都是在多线程环境中使用，所以要注意并发冲突问题。注意在run方法中，我们没有使用synchronized代码块或者Lock来解决并发问题，而是使用了CAS这个乐观锁来实现并发安全，保证只有一个线程能运行FutureTask任务。

get方法

get方法就是阻塞获取线程执行结果，这里了主要做了两件事情

1.判断当前的状态，如果状态小于等于COMPLETING,表示FutureTask任务还没有完结，所以调用awaitDone方法，让当前线程等待。

2.report返回结果值或者抛出异常

public V get() throws InterruptedException, ExecutionException {
  int s = state;
  if (s <= COMPLETING)
    s = awaitDone(false, 0L);
  return report(s);
}

awaitDone

如果当前的结果还没有被执行完，把当前线程插入到等待队列

private int awaitDone(boolean timed, long nanos)
  throws InterruptedException {
  final long deadline = timed ? System.nanoTime() + nanos : 0L;
  WaitNode q = null;
  boolean queued = false;//节点是否已添加
  for (;;) {
    //如果当前线程中断标志位是true,
    //那么从队列中移除节点q,并抛出InterruptedException异常
    if (Thread.interrupted()) {
      removeWaiter(q);
      throw new InterruptedException();
    }

    int s = state;
    //当状态大于COMPLETING时，表示Future任务已结束。
    if (s > COMPLETING) {
      if (q != null)
        //将节点q线程设置为null,因为线程没有阻塞等待
        q.thread = null;
      return s;
    }//表示还有一些后序操作没有完成，那么当前线程让出执行权
    else if (s == COMPLETING) // cannot time out yet
      Thread.yield();
    //表示状态是new,那么就需要将当前线程阻塞等待。
    //就是将它插入等待线程链表中
    else if (q == null)
      q = new WaitNode();
    else if (!queued)
      //使用CAS函数将节点添加到链表中，如果添加失败，那么queued为false,
      //下次循环时，会继续添加，直到成功
      queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                           q.next = waiters, q);
    else if (timed) {//time为true表示需要设置超时
      nanos = deadline - System.nanoTime();
      if (nanos <= 0L) {
        removeWaiter(q);
        return state;
      }
      LockSupport.parkNanos(this, nanos);//让当前线程等待nanos时间
    }
    else
      LockSupport.park(this);
  }
}

被阻塞的线程，会等到run方法执行结束之后被唤醒

report

report方法就是根据传入的状态值，来决定是抛出异常，还是返回结果值。这个两种情况都表示FutureTask完结了。

private V report(int s) throws ExecutionException {
  Object x = outcome;//表示call的返回值
  if (s == NORMAL)//表示正常完结状态，所以返回结果值
    return (V)x;
  //大于或等于CANCELLED,都表示手动取消FutureTask任务。
  //所以抛出CancellationExecption异常
  if (s >= CANCELLED)
    throw new CancellationException();
  //否则就是运行过程中，发生了异常，这里就抛出这个异常
  throw new ExecutionException((Throwable)x);
}

线程池对于Future/Callable的执行

现在看线程池里面的submit方法，就很清楚了。

public static void main(String[] args) throws ExecutionException, InterruptedException {
  ExecutorService executorService = Executors.newFixedThreadPool(1);
  CallableDemo callableDemo = new CallableDemo();
  Future<String> future = executorService.submit(callableDemo);
  System.out.println(future.get());
}

AbstractExecutorService.submit

调用抽象类中的submit方法，这里其实相等对于execute方法来说，只多做了一步操作，就是封装了一个RunnableFuture

public <T> Future<T> submit(Callable<T> task) {
  if (task == null) throw new NullPointerException();
  RunnableFuture<T> ftask = newTaskFor(task);
  execute(ftask);
  return ftask;
}

ThreadpoolExecutor.execute

然后调用execute方法，这里面的逻辑前面分析过了。会通过worker线程来调用ftask的run方法。而这个ftask其实就是 FutureTask 里面最终实现的逻辑。