java线程池源码分析

在什么情况下使用线程池

1. 单个任务处理的时间比较短
2. 将需处理的任务的数量大

使用线程池的好处

1. 减少在创建和销毁线程上所花的时间以及系统资源的开销
2. 如不使用线程池，有可能造成系统创建大量线程而导致消耗完系统内存以及”过度切换”

案例1

public class Main {
    private static Executor executor = Executors.newFixedThreadPool(10);
    public static void main(String[] args){
        for(int i = 0; i < 20; i++){
            executor.execute(new Task());
        }
    }
    private static class Task implements Runnable{
        public void run(){
            System.out.println(Thread.currentThread().getName());
        }
    }
}

1. Executor.newFixedThreadPool(10)初始化一个包含10个线程的executor;
2. 通过executor.execute方法提交20个任务
3. 负责执行任务的生命周期都由Executor框架进行管理

ThreadPoolExecutor

Executor是java线程池的工厂类**，通过它可以快速初始化一个符合业务需求的线程池，如Excutor.newFixedThreadPool方法可以生成一个拥有固定线程数的线程池。

/**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters and default thread factory and rejected execution handler.
 * It may be more convenient to use one of the {@link Executors} factory
 * methods instead of this general purpose constructor.
 *
 * @param corePoolSize the number of threads to keep in the pool, even
 *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
 * @param maximumPoolSize the maximum number of threads to allow in the
 *        pool
 * @param keepAliveTime when the number of threads is greater than
 *        the core, this is the maximum time that excess idle threads
 *        will wait for new tasks before terminating.
 * @param unit the time unit for the {@code keepAliveTime} argument
 * @param workQueue the queue to use for holding tasks before they are
 *        executed.  This queue will hold only the {@code Runnable}
 *        tasks submitted by the {@code execute} method.
 * @throws IllegalArgumentException if one of the following holds:<br>
 *         {@code corePoolSize < 0}<br>
 *         {@code keepAliveTime < 0}<br>
 *         {@code maximumPoolSize <= 0}<br>
 *         {@code maximumPoolSize < corePoolSize}
 * @throws NullPointerException if {@code workQueue} is null
 */
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

其本质是通过不同的参数初始化一个ThreadPoolExecutor对象，具体参数描述如下：

corePoolSize

线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize；如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行；如果执行了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有核心线程。

maximumPoolSize

线程池中允许的最大线程数。如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize。

keepAliveTime

线程空闲时的存活时间，即当线程没有任务执行时，继续存活的时间；默认情况下，该参数只在线程数大于corePoolSize时才有用。

unit

keepAliveTime的单位。

workQueue

用来保存等待被执行的任务的阻塞队列，且任务必须实现Runable接口，在JDK中提供了如下阻塞队列：

1. ArrayBlockingQueue：基于数组结构的有界阻塞队列，按FIFO排序任务；
2. LinkedBlockingQuene：基于链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQuene；
3. SynchronousQuene：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQuene；
4. priorityBlockingQuene：具有优先级的无界阻塞队列；

threadFactory

创建线程的工厂，通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名。

DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

handler

线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略：

1. AbortPolicy：直接抛出异常，默认策略；
2. CallerRunsPolicy：用调用者所在的线程来执行任务；
3. DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
4. DiscardPolicy：直接丢弃任务；

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

Exectors

Exectors工厂类提供了线程池的初始化接口，主要有如下几种:

newFixedThreadPool

初始化一个指定线程数的线程池，其中corePoolSize == maximumPoolSize，使用LinkedBlockingQuene作为阻塞队列，不过当线程池没有可执行任务时，也不会释放线程。

/**
    * Creates a thread pool that reuses a fixed number of threads
    * operating off a shared unbounded queue.  At any point, at most
    * {@code nThreads} threads will be active processing tasks.
    * If additional tasks are submitted when all threads are active,
    * they will wait in the queue until a thread is available.
    * If any thread terminates due to a failure during execution
    * prior to shutdown, a new one will take its place if needed to
    * execute subsequent tasks.  The threads in the pool will exist
    * until it is explicitly {@link ExecutorService#shutdown shutdown}.
    *
    * @param nThreads the number of threads in the pool
    * @return the newly created thread pool
    * @throws IllegalArgumentException if {@code nThreads <= 0}
    */
   public static ExecutorService newFixedThreadPool(int nThreads) {
       return new ThreadPoolExecutor(nThreads, nThreads,
                                     0L, TimeUnit.MILLISECONDS,
                                     new LinkedBlockingQueue<Runnable>());
   }

newCachedThreadPool

1. 初始化一个可以缓存线程的线程池，默认缓存60s，线程池的线程数可达到Integer.MAX_VALUE，即2147483647，内部使用SynchronousQueue作为阻塞队列；
2. 和newFixedThreadPool创建的线程池不同，newCachedThreadPool在没有任务执行时，当线程的空闲时间超过keepAliveTime，会自动释放线程资源，当提交新任务时，如果没有空闲线程，则创建新线程执行任务，会导致一定的系统开销;

/**
  * Creates a thread pool that creates new threads as needed, but
  * will reuse previously constructed threads when they are
  * available.  These pools will typically improve the performance
  * of programs that execute many short-lived asynchronous tasks.
  * Calls to {@code execute} will reuse previously constructed
  * threads if available. If no existing thread is available, a new
  * thread will be created and added to the pool. Threads that have
  * not been used for sixty seconds are terminated and removed from
  * the cache. Thus, a pool that remains idle for long enough will
  * not consume any resources. Note that pools with similar
  * properties but different details (for example, timeout parameters)
  * may be created using {@link ThreadPoolExecutor} constructors.
  *
  * @return the newly created thread pool
  */
 public static ExecutorService newCachedThreadPool() {
     return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                   60L, TimeUnit.SECONDS,
                                   new SynchronousQueue<Runnable>());
 }

Java线程池的原理及几类线程池的介绍

深入分析java线程池的实现原理