Java线程池--原理及源码分析

拼搏现实的明天。 2022-07-15 00:23 146阅读 0赞

http://www.jianshu.com/p/117571856b28

合理使用线程池能够带来3个好处。

第一，降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。  
第二，提高响应速度。当任务到达时，任务可以不需要等到线程创建就立即执行。  
第三，提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以统一的分配、调优和监控。

##### 以下是本文的目录大纲： #####

*  ThreadPoolExecutor类
 *  线程池实现原理和源码分析
 *  使用示例
 *  合理配置线程池大小

##### 一、ThreadPoolExecutor类 #####

java.util.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类，该类提供4个构造方法：

public class ThreadPoolExecutor extends AbstractExecutorService {
    ......
        public ThreadPoolExecutor(int corePoolSize,
                                      int maximumPoolSize,
                                      long keepAliveTime,
                                      TimeUnit unit,
                                      BlockingQueue<Runnable> workQueue) {
                this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                    Executors.defaultThreadFactory(), defaultHandler);
        }
    
        public ThreadPoolExecutor(int corePoolSize,
                                      int maximumPoolSize,
                                      long keepAliveTime,
                                      TimeUnit unit,
                                      BlockingQueue<Runnable> workQueue,
                                      ThreadFactory threadFactory) {
                this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                     threadFactory, defaultHandler);
        }
    
        public ThreadPoolExecutor(int corePoolSize,
                                      int maximumPoolSize,
                                      long keepAliveTime,
                                      TimeUnit unit,
                                      BlockingQueue<Runnable> workQueue,
                                      RejectedExecutionHandler handler) {
                this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                     Executors.defaultThreadFactory(), handler);
        }
    
        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.corePoolSize = corePoolSize;
                this.maximumPoolSize = maximumPoolSize;
                this.workQueue = workQueue;
                this.keepAliveTime = unit.toNanos(keepAliveTime);
                this.threadFactory = threadFactory;
                this.handler = handler;
        }

从上面代码可知，ThreadPoolExecutor继承了AbstractExecutorService，而且，前三个构造器均调用第四个构造器进行初始化工作。

下面解释下构造器中各个参数的含义：

*  **corePoolSize** ：核心池的大小，该参数与之后的线程池实现原理有很大的关系。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用prestartAllCoreThreads()和prestartCoreThread()方法，从方法名字可以看出，是预创建线程的意思，即在没有任务到来之前，就创建corePoolSize个线程或1个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
 *  **maximumPoolSize** ：线程池中的最大线程数，表示线程池中最多能创建多少个线程。
 *  **keepAliveTime** ：表示线程没有任务执行时最多存活多久。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，知道线程池中的线程不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean value)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0；
 *  **unit** ：参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性：

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

*  **workQueue** ： 一个阻塞队列，用来存储等待执行的任务。该参数也很重要，会对线程池的运行过程产生巨大影响，一般而言，有一下几种选择：

> ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序；
> 
> LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列；
> 
> SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列；
> 
> PriorityBlockingQueue：一个具有优先级的无限阻塞队列；

*  **threadFactory** ：线程工厂，主要用于创建线程；
 *  **handler** ：饱和策略，即当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。以下是JDK1.5提供的四种策略：

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

从上面给出的ThreadPoolExecutor类的代码可以知道，ThreadPoolExecutor继承了AbstractExecutorService，我们来看一下AbstractExecutorService的实现：

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 {
        };
    
    }

AbstractExecutorService是一个抽象类，它实现了ExecutorService接口。

我们接着看ExecutorService接口的实现：

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;
    }

而ExecutorService又继承了Executor接口，实现如下：

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

到这里，大家应该明白了ThreadPoolExecutor、AbstractExecutorService、ExecutorService和Executor几个之间的关系了。

Executor是一个顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型，从字面意思可以理解，就是用来执行传进去的任务的；

然后ExecutorService接口继承了Executor接口，并声明了一些方法：submit、invokeAll、invokeAny以及shutDown等；

抽象类AbstractExecutorService实现了ExecutorService接口，基本实现了ExecutorService中声明的所有方法；

然后ThreadPoolExecutor继承了类AbstractExecutorService。

在ThreadPoolExecutor类中有几个非常重要的方法：

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

execute()方法实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。

submit()方法是在ExecutorService中声明的方法，在AbstractExecutorService就已经有了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果。

shutdown()和shutdownNow()是用来关闭线程池的。

还有很多其他的方法：

比如：getQueue() 、getPoolSize() 、getActiveCount()、getCompletedTaskCount()等获取与线程池相关属性的方法，有兴趣的朋友可以自行查阅API。

##### 二、线程池实现原理 #####

下面将深入解析一下线程池的实现原理，分为以下几个方面：

1.  线程池状态
2.  任务的执行
3.  线程池中的线程初始化
4.  任务缓存队列及排队策略
5.  任务拒绝策略
6.  线程池的关闭
7.  线程池容量的动态调整

###### 1. 线程池状态 ######

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

当创建线程池后，初始时，线程池处于RUNNING状态；

如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时线程池不能够接受新的任务，它会等待所有任务执行完毕；

如果调用了shutdownNow()方法，则线程池处于STOP状态，此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务；

当线程池处于SHUTDOWN或STOP状态，并且所有工作线程已经销毁，任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATED状态。

###### 2. 任务的执行 ######

在了解将任务提交给线程池到任务执行完毕整个过程之前，我们先来看一下ThreadPoolExecutor类中其他的一些比较重要成员变量：

> private final BlockingQueue<Runnable> workQueue; // 任务缓存队列，用来存放等待执行的任务
> 
> private final ReentrantLock mainLock = new ReentrantLock(); // 线程池的主要状态锁，对线程池的状态（比如线程池的大小，状态等）的改变都要使用这个锁
> 
> 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; //用来记录已经执行完毕的任务个数

每个变量的作用都已经标明出来了，这里要重点解释一下corePoolSize、maximumPoolSize、largestPoolSize三个变量。

corePoolSize在很多地方被翻译成核心池大小，其实我的理解这个就是线程池的大小。举个简单的例子：

假如有一个工厂，工厂里面有10个工人，每个工人同时只能做一件任务。

因此只要当10个工人中有工人是空闲的，来了任务就分配给空闲的工人做；

当10个工人都有任务在做时，如果还来了任务，就把任务进行排队等待；

如果说新任务数目增长的速度远远大于工人做任务的速度，那么此时工厂主管可能会想补救措施，比如重新招4个临时工人进来；

然后就将任务也分配给这4个临时工人做；

如果说着14个工人做任务的速度还是不够，此时工厂主管可能就要考虑不再接收新的任务或者抛弃前面的一些任务了。

当这14个工人当中有人空闲时，而新任务增长的速度又比较缓慢，工厂主管可能就考虑辞掉4个临时工了，只保持原来的10个工人，毕竟请额外的工人是要花钱的。

这个例子中的corePoolSize就是10，而maximumPoolSize就是14（10+4）。

也就是说corePoolSize就是线程池大小，maximumPoolSize在我看来是线程池的一种补救措施，即任务量突然过大时的一种补救措施。

不过为了方便理解，在本文后面还是将corePoolSize翻译成核心池大小。

largestPoolSize只是一个用来起记录作用的变量，用来记录线程池中曾经有过的最大线程数目，跟线程池的容量没有任何关系。不过，在分析问题时，可以知道线程池是否满过。

下面我们进入正题，看一下任务从提交到最终执行完毕经历了哪些过程。

在ThreadPoolExecutor类中，最核心的任务提交方法是execute()方法，虽然通过submit也可以提交任务，但是实际上submit方法里面最终调用的还是execute()方法，所以我们只需要研究execute()方法的实现原理即可：

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            /*
             * Proceed in 3 steps:
             *
             * 1. If fewer than corePoolSize threads are running, try to
             * start a new thread with the given command as its first
             * task.  The call to addWorker atomically checks runState and
             * workerCount, and so prevents false alarms that would add
             * threads when it shouldn't, by returning false.
             *
             * 2. If a task can be successfully queued, then we still need
             * to double-check whether we should have added a thread
             * (because existing ones died since last checking) or that
             * the pool shut down since entry into this method. So we
             * recheck state and if necessary roll back the enqueuing if
             * stopped, or start a new thread if there are none.
             *
             * 3. If we cannot queue task, then we try to add a new
             * thread.  If it fails, we know we are shut down or saturated
             * and so reject the task.
             */
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) {
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            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);
        }