菜鸟谈——线程池初探

秒速五厘米 2024-02-18 18:06 82阅读 0赞

## 菜鸟谈——线程池初探（jdk1.8） ##

### 1. 前言 ###

如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建线程就会大大降低系统的效率，因为频繁创建线程和销毁线程需要时间。

java中可以通过线程池来达到这样的效果。

### 2. Java中ThreadPoolExecutor类 ###

#### 2.1 构造方法 ####

// Public constructors and methods
        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);
        }
    
        /**
         * Creates a new {@code ThreadPoolExecutor} with the given initial
         * parameters.
         *
         * @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.
         * @param threadFactory the factory to use when the executor
         *        creates a new thread
         * @param handler the handler to use when execution is blocked
         *        because the thread bounds and queue capacities are reached
         * @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}
         *         or {@code threadFactory} or {@code handler} is null
         */
        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;
        }

有四个构造器，其中前三个构造器都是调用了第四个构造器进行的初始化工作。

构造器中各个参数的含义：

**int corePoolSize ：** 核心线程池的大小。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartCoreThread()和prestartAllCoreThreads()方法，这两个函数是预创建线程的意思，即在没有到来任务之前，创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数是0，当有任务来之后，就会创建线程去执行任务，当线程池中的线程数目达到corePoolSizehou ，就会把到达的任务放到缓存队列当中。

**int maximumPoolSize：**线程池最大大小，表示在线程池中最多创建多少个线程。

**long keepAliveTime：**保持活动时间。线程没有任务时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，即当线程池中的线程数大于corePoolSize时，如果一个线程的空闲时间达到keepAliveTime，则会终止，直到线程池综合功能的线程数不超过corePoolSize。

但是如果调用了allowCoreThreadTimeOut(boolean)方法，keepAliveTime参数才会起作用，直到线程池中的线程数为0。

**TimeUnit：**参数keepAliveTime的时间单位，有7中取值

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

**BlockingQueue：**一个阻塞队列，用来存储等待执行的任务，有以下几种选择：

ArrayBlockingQueue<E> arrayBlockingQueue = null;
    LinkedBlockingDeque<E> blockingDeque = null;
    SynchronousQueue<E> es = null;

**ThreadFactory：**线程工厂，用来创建线程

**RejectedExecutionHandler：**驳回回调

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

#### 2.2 类关系 ####

public class ThreadPoolExecutor extends AbstractExecutorService {
        }

ThreadPoolExecutor继承了AbstractExecutorService类，

AbstractExecutorService实现了ExecutorService的接口，

ExecutorService继承于Executor。

##### 2.2.1 Executor类 #####

Executor是一个顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型。。即用来执行传进去的任务的；

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

##### 2.2.2 ExecutorService类 #####

然后ExecutorService接口继承了Executor接口，并声明了一些方法submit，invokeAll，invokeAny，shutdown，shutdownNow，isShutdown，isTerminated：

public interface ExecutorService extends Executor {
        
        void shutdown();
        //关闭线程池的操作，调用此方法后线程池不再接收新的任务，但是会把当前缓存队列中的任务执行完毕。
    
        List<Runnable> shutdownNow();
        //关闭线程池的操作，调用此方法后，不但不能接受新的任务，也会尝试中断正在执行的任务，同时不再执行缓存队列中的任务
    
        boolean isShutdown();
        boolean isTerminated();
        boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException;
    
        <T> Future<T> submit(Callable<T> task);
        //接受callable参数，增加了返回值参数Future
    
        <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;
        //invokeAll提供一系列的任务的功能，最后返回所有任务的Future对象，用于获取任务的执行结果
    
        <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;
    }

##### 2.2.3 抽象类AbstractExecutorService #####

抽象类AbstractExecutorService实现了ExecutorService接口，实现了ExecutorService中大部分的接口。

public <T> Future<T> submit(Callable<T> task) {
      if (task == null) throw new NullPointerException();
      RunnableFuture<T> ftask = newTaskFor(task);
      execute(ftask);
      return ftask;
    }
    //解析：任务为null 抛出空指针异常，新建一个ftask对象，然后还是调用了execute方法去执行任务
    
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
      return new FutureTask<T>(callable);
    }
    //解析：返回FutureTask对象，这个对象既实现了Runnable接口又实现了Callable接口，这样既可以当做线程的执行对象又可以对人物执行后的结果进行获取。

public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException {
            if (tasks == null)
                throw new NullPointerException();
            //创建存储结果的集合futures，大小为传入的任务数（一个任务对应一个future对象）
            ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
            boolean done = false;
            try {
                //遍历所有的Callable对象           
                for (Callable<T> t : tasks) {
                    //把他们封装到RunnableFuture中（实际传入的是FutureTask对象）
                    RunnableFuture<T> f = newTaskFor(t);
                    //然后把创建的futureTask对象加入到结果集futures中
                    futures.add(f);
                    //然后调用execute执行传入的任务
                    execute(f);
                }
                //for循环保证每个任务已经执行完毕
                for (int i = 0, size = futures.size(); i < size; i++) {
                    //获取任务
                    Future<T> f = futures.get(i);
                    //判断任务是否已经执行完成
                    if (!f.isDone()) {
                        try {
                            //任务没有执行完成，调用get方法，阻塞，当前线程等待任务执行结束才返回。
                            f.get();
                        } catch (CancellationException ignore) {
                        } catch (ExecutionException ignore) {
                        }
                    }
                }
                done = true;
                //返回结果集
                return futures;
            } finally {
                if (!done)
                    for (int i = 0, size = futures.size(); i < size; i++)
                        futures.get(i).cancel(true);
            }
        }

##### 2.2.4 ThreadPoolExecutor类 #####

然后ThreadPoolExecutor继承了类AbstractExecutorService，在ThreadPoolExecutor有几个比较重要的方法：

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

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

**具体在任务执行中讲解 3.2**

### 3. 深入剖析线程池实现原理 ###

#### 3.1 线程池状态 ####

可状态有一些几种

*   RUNNING:  Accept new tasks and process queued tasks
         *   SHUTDOWN: Don't accept new tasks, but process queued tasks
         *   STOP:     Don't accept new tasks, don't process queued tasks,
         *             and interrupt in-progress tasks
         *   TIDYING:  All tasks have terminated, workerCount is zero,
         *             the thread transitioning to state TIDYING
         *             will run the terminated() hook method
         *   TERMINATED: terminated() has completed
        // runState is stored in the high-order bits
        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;
        private static final int TERMINATED =  3 << COUNT_BITS;

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

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

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

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

TIDYING ？？？？？？

#### 3.2 任务的执行 ####

重要的成员变量

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

着重解析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();
            //workerCountOf(c)就是计算当前线程池中的线程数，如果小于核心线程数
            if (workerCountOf(c) < corePoolSize) {
                //如果添加成功，那么方法执行结束
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            //判断线程池是否处于运行状态【isRunning(c)】，
            //且尝试往队列中添加任务【workQueue.offer(command)】
            if (isRunning(c) && workQueue.offer(command)) {
                int recheck = ctl.get();
                //考虑多线程问题，需再次进行double-check
                if (! isRunning(recheck) && remove(command))
                    reject(command); //拒绝任务
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            //如果尝试添加缓存队列失败，那么就尝试添加新的线程
            else if (!addWorker(command, false))
                reject(command);//拒绝任务
        }

#### 3.3 线程池中的线程初始化 ####

默认情况下，创建线程之后，线程池中是没有线程的，需要提交任务后才会创建线程。

在实际中，如果需要线程池创建之后立即创建线程，可以通过下面两个方法办到：

//初始化一个核心线程
    public boolean prestartCoreThread() {
      return workerCountOf(ctl.get()) < corePoolSize &&
        addWorker(null, true); //注意：第一个参数传入null
    }
    
    //初始化所有核心线程
    public int prestartAllCoreThreads() {
      int n = 0;
      while (addWorker(null, true)) //注意：第一个参数传入null
        ++n;
      return n;
    }

#### 3.4 任务缓存队列及排队策略 ####

任务缓存队列 BlockingQueue，通常有三种类型：

ArrayBlockingQueue<E> arrayBlockingQueue = null; //基于数组的先进先出队列，此队列创建时必须指定大小
    LinkedBlockingDeque<E> blockingDeque = null;//基于链表的先进先出队列，如果创建时没有执行此队列的大小，则默认为Integer.MAX_VALUE；
    SynchronousQueue<E> es = null; //特殊，不会保存提交的任务，而是直接创建一个线程来执行新来的任务

#### 3.5 任务拒绝策略 ####

当线程池的任务缓存队列已经满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来，就会采取任务拒绝策略，通常有4种策略如下：

#### 3.6 线程池的关闭 ####

ThreadPoolExecutor提供了两个方法，用于线程池的关闭，分别是shutdown()和shutdownNow()

void shutdown();
        //关闭线程池的操作，调用此方法后线程池不再接收新的任务，但是会把当前缓存队列中的任务执行完毕。
    
        List<Runnable> shutdownNow();
        //关闭线程池的操作，调用此方法后，不但不能接受新的任务，也会尝试中断正在执行的任务，同时不再执行缓存队列中的任务

#### 3.7 线程池容量的动态调整 ####

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

setCorePoolSize()：设置核心池大小

setMaximumPoolSize()：设置线程池最大能创建的线程数目大小