Java线程池ThreadPoolExecutor的使用及其原理

矫情吗；* 2021-06-10 20:41 570阅读 0赞

### 文章目录 ###

*   *   *  介绍
         *   *  什么是线程池
             *  线程池的作用
         *  关于Java线程池（ThreadPool)
         *   *  jdk提供的线程池有哪些
             *   *  查看Executor源码会发现，Executor只是一个创建线程池的工具类，这四种方式创建的源码就会发现，都是利用 ThreadPoolExecutor 类实现的,真正的线程池接口是ExecutorService,其实现类为ThreadPoolExecutor。
             *  关于创建线程池
             *  通常我们使用线程池可以通过如下方式去使用：
             *   *  使用jdk提供的线程池
                 *  创建自定义线程池
             *  线程池的任务队列
             *  线程池的任务拒绝策略
             *  线程池如何执行
             *   *  主要分三步进行：
             *  线程池的任务添加流程
             *  线程池中任务的执行流程
             *  如何关闭线程池

### 介绍 ###

#### 什么是线程池 ####

> 线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。如果某个线程在托管代码中空闲（如正在等待某个事件）,则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙，但队列中包含挂起的工作，则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队，但他们要等到其他线程完成后才启动。

#### 线程池的作用 ####

*  提高程序运行效率：创建一个新线程需要消耗一定的时间，而线程中的线程在使用时已经创建完成，可节省此部分开销。
 *  解耦作用；线程的创建与执行完全分开，方便维护。
 *  资源复用：每次线程执行完毕后会重新放入线程池中供其它调用者使用,可以减少线程销毁的开销。
 *  提高cpu利用率：线程数过多会导致系统内核额外的线程调度开销，线程池可以控制创建线程数量以实现性能最大化。线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量，默认情况下线程数一般取cpu数量+2比较合适。
 *  系统健壮性：接受突发性的大量请求，不至于使服务器因此产生大量线程去处理请求而导致崩溃。

### 关于Java线程池（ThreadPool) ###

#### jdk提供的线程池有哪些 ####

在 JDK 1.5 之后推出了相关的 api，其提供的线程池有以下四种：

*  Executors.newCachedThreadPool()：创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，那么就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。
 *  Executors.newFixedThreadPool(nThreads)：创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。
 *  Executors.newSingleThreadExecutor()：创建单个线程的线程池。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
 *  Executors.newScheduledThreadPool(corePoolSize)：创建一个大小无限的线程池。此线程池支持定时以及周期性执行任务的需求。

##### 查看Executor源码会发现，Executor只是一个创建线程池的工具类，这四种方式创建的源码就会发现，都是利用 ThreadPoolExecutor 类实现的,真正的线程池接口是ExecutorService,其实现类为ThreadPoolExecutor。 #####

**Executors** 部分源码：

public class Executors {
    
    	public static ExecutorService newFixedThreadPool(int nThreads) {
            	return new ThreadPoolExecutor(nThreads, nThreads,
                                 	         0L, TimeUnit.MILLISECONDS,
                                	          new LinkedBlockingQueue<Runnable>());
      	  }
        public static ExecutorService newCachedThreadPool() {
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                          60L, TimeUnit.SECONDS,
                                          new SynchronousQueue<Runnable>());
        }
        public static ExecutorService newSingleThreadExecutor() {
            return new FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1,
                                        0L, TimeUnit.MILLISECONDS,
                                        new LinkedBlockingQueue<Runnable>()));
        }
        public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
            return new ScheduledThreadPoolExecutor(corePoolSize);
        }

**ThreadPoolExecutor** 最终的构造器源码：

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

*  corePoolSize：线程池的大小。线程池创建之后不会立即去创建线程，而是等待线程的到来。当前执行的线程数大于该值时，线程会加入到缓冲队列。
 *  maximumPoolSize：线程池中创建的最大线程数。
 *  keepAliveTime：空闲的线程多久时间后被销毁。默认情况下，该值在线程数大于corePoolSize时，对超出- corePoolSize值的这些线程起作用。
 *  unit：TimeUnit枚举类型的值，代表keepAliveTime时间单位。
 *  workQueue： 用于在执行任务之前保存任务的队列。此队列将仅包含execute方法提交的可运行任务。
 *  threadFactory：执行程序创建新线程时要使用的工厂
 *  handler：线程拒绝策略。 当队列和最大线程池都满了之后的饱和策略。
    
    **其它构造器如下：**
    
        // 构造器一
            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);
            }

#### 关于创建线程池 ####

有一点是肯定的，线程池肯定是不是越大越好。  
通常我们是需要根据这批任务执行的性质来确定的。

*  IO 密集型任务：由于线程并不是一直在运行，所以可以尽可能的多配置线程，比如 CPU 个数 \* 2
 *  CPU 密集型任务（大量复杂的运算）应当分配较少的线程，比如 CPU 个数相当的大小。

当然这些都是经验值，最好的方式还是根据实际情况测试得出最佳配置。

#### 通常我们使用线程池可以通过如下方式去使用： ####

##### 使用jdk提供的线程池 #####

//创建一个可缓存的线程池
            ExecutorService cachedThreadPool= Executors.newCachedThreadPool();
            //通过execute方法执行接口
            cachedThreadPool.execute(()->{
                //Runnable to do something.
            });

##### 创建自定义线程池 #####

ExecutorService threadPool = new ThreadPoolExecutor(// 自定义一个线程池
                    1, // 核心线程数
                    2, // 最大线程数
                    60, // 超过核心线程数的额外线程存活时间
                    TimeUnit.SECONDS, // 线程存活时间的时间单位
                    new ArrayBlockingQueue<>(3) // 有界队列，容量是3个
                    , Executors.defaultThreadFactory()    // 线程工厂
                    , new ThreadPoolExecutor.AbortPolicy() //线程的拒绝策略
            );
            //执行一个任务
            threadPool.execute(()->{
                //线程执行的具体逻辑
                //Runnable to do something.
            });

#### 线程池的任务队列 ####

队列有三种通用策略：

*  直接提交。工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
 *  无界队列。使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
 *  有界队列。当使用有限的 maximumPoolSizes时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

#### 线程池的任务拒绝策略 ####

jdk提供了四种拒绝策略：

*  AbortPolicy： 丢弃任务，并抛出拒绝执行 RejectedExecutionException 异常信息。必须处理好抛出的异常，否则会打断当前的执行流程，影响后续的任务执行。同时它也是线程池默认的拒绝策略。
 *  CallerRunsPolicy： 当触发拒绝策略，只要线程池没有关闭的话，则使用调用线程直接运行任务。一般并发比较小，性能要求不高，不允许失败。但是，由于调用者自己运行任务，如果任务提交速度过快，可能导致程序阻塞，性能效率上必然的损失较大
 *  DiscardPolicy： 直接丢弃，其他啥都没有
 *  DiscardOldestPolicy： 当触发拒绝策略，只要线程池没有关闭的话，丢弃阻塞队列 workQueue 中最老的一个任务，并将新任务加入

#### 线程池如何执行 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2MxNTE1ODAzMjMxOQ_size_16_color_FFFFFF_t_70_pic_center]

**所有任务的调度都是由execute方法完成的**，这部分完成的工作是：检查现在线程池的**运行状态、运行线程数、运行策略**，决定接下来执行的流程，是直接申请线程执行，或是缓冲到队列中执行，亦或是直接拒绝该任务。

##### 主要分三步进行： #####

1.  如果运行的线程少于corePoolSize，请尝试以给定的命令作为第一个线程开始一个新线程任务。对addWorke的调用以原子方式检查运行状态和workerCount，从而防止会增加当它不应该的时候，返回false。
2.  如果任务可以成功排队，那么我们仍然需要仔细检查是否应该添加线程（因为自从上次检查以来已有的已经死了）或者自进入此方法后，池已关闭。所以我们重新检查状态，必要时回滚排队已停止，如果没有，则启动新线程。
3.  如果无法将任务排队，则尝试添加新的线程。如果失败了，我们就知道我们已经关闭或者饱和了，所以拒绝这个任务。

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
    
            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);
        }

#### 线程池的任务添加流程 ####

在excute方法中调用了addWorker方法

private boolean addWorker(Runnable firstTask, boolean core) {
         	// 外层循环，负责判断线程池状态，处理线程池状态变量加1操作
            retry:
            for (;;) {
    		    // 状态总体相关值：运行状态 + 执行线程任务数量 
                int c = ctl.get();
    			// 读取状态值 - 运行状态
                int rs = runStateOf(c);
                // Check if queue empty only if necessary.
    			// 满足下面两大条件的，说明线程池不能接受任务了，直接返回false处理
    			// 主要目的就是想说，只有线程池的状态为 RUNNING 状态时，线程池才会接收
    			// 新的任务，增加新的Worker工作线程
    			// 线程池的状态已经至少已经处于不能接收任务的状态了
                if (rs >= SHUTDOWN &&
    			  //目的是检查线 程池是否处于关闭状态
                    ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
                           return false;
    				// 内层循环，负责worker数量加1操作
    				for (;;) {
    				// 获取当前worker线程数量
    					int wc = workerCountOf(c);
    					if (wc >= CAPACITY ||
    						// 如果线程池数量达到最大上限值CAPACITY
    						// core为true时判断是否大于corePoolSize核心线程数量
    						// core为false时判断是否大于maximumPoolSize最大设置的线程数量
    						wc >= (core ? corePoolSize : maximumPoolSize))
    						return false;
    					// 调用CAS原子操作，目的是worker线程数量加1
    					if (compareAndIncrementWorkerCount(c)) //
    						break retry;
    					c = ctl.get();  // Re-read ctl
    					// CAS原子操作失败的话，则再次读取ctl值
    					if (runStateOf(c) != rs)
    					// 如果刚刚读取的c状态不等于先前读取的rs状态，则继续外层循环判断
    						continue retry;
    					// else CAS failed due to workerCount change; retry inner loop
    					// 之所以会CAS操作失败，主要是由于多线程并发操作，导致workerCount
    					// 工作线程数量改变而导致的，因此继续内层循环尝试操作
    				}
            }
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try {
    			// 创建一个Worker工作线程对象，将任务firstTask，
    			// 新创建的线程thread都封装到了Worker对象里面
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) {
    				// 由于对工作线程集合workers的添加或者删除，
    				// 涉及到线程安全问题，所以才加上锁且该锁为非公平锁
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try {
                        // Recheck while holding lock.
                        // Back out on ThreadFactory failure or if
                        // shut down before lock acquired.
    					// 获取锁成功后，执行临界区代码，首先检查获取当前线程池的状态rs
                        int rs = runStateOf(ctl.get());
    					// 当线程池处于可接收任务状态
    					// 或者是不可接收任务状态，但是有可能该任务等待队列中的任务
    					// 满足这两种条件时，都可以添加新的工作线程
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                             workers.add(w);
    						// 添加新的工作线程到工作线程集合workers，workers是set集合
                            int s = workers.size();
                            if (s > largestPoolSize) 
    						// 变量记录了线程池在整个生命周期中曾经出现的最大线程个数
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally {
                        mainLock.unlock();
                    }
                    if (workerAdded) { 
    				// 往workers工作线程集合中添加成功后，则立马调用线程start方法启动起来
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
                if (! workerStarted)
    			// 如果启动线程失败的话，还得将刚刚添加成功的线程共集合中移除并且做线				// 程数量做减1操作
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

#### 线程池中任务的执行流程 ####

**runWorker** **通过调用t.start()启动了线程，线程池真正核心执行任务的地方就在此**

final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
    		// allow interrupts 允许中断
            w.unlock();
            boolean completedAbruptly = true;
            try {
    			// 不断从等待队列blockingQueue中获取任务
    			// 之前addWorker(null, false)这样的线程执行时，
    			// 会通过getTask中再次获取任务并执行
                while (task != null || (task = getTask()) != null) {
                    w.lock(); 
    				// 上锁，并不是防止并发执行任务，
    				// 而是为了防止shutdown()被调用时不终止正在运行的worker线程
                    // 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 {
    					// task.run()执行前，由子类实现
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try {
                            task.run(); // 执行线程Runable的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 {
    						// task.run()执行后，由子类实现
                            afterExecute(task, thrown);
                        }
                    } finally {
                        task = null;
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally {
                processWorkerExit(w, completedAbruptly);
            }
        }

#### 如何关闭线程池 ####

有运行任务自然也有关闭任务。  
通过查看ExecutorService接口，其实无非就是两个方法 shutdown()/shutdownNow()。  
但他们有着重要的区别：

*  shutdown() 执行后停止接受新任务，会把队列的任务执行完毕。
 *  shutdownNow() 也是停止接受新任务，但会中断所有的任务，将线程池状态变为 stop。

threadPool.shutdown();
            threadPool.shutdownNow();

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2MxNTE1ODAzMjMxOQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20210606/778c6e1a16ae4dbab7af3574a399733e.png