四张图看懂线程池

Posted 2021-04-23 迷茫的码农

四张图看懂线程池

程序的运行，本质上是对系统资源(CPU、内存、磁盘、网络等)的使用。如何高效的使用这些资源，是编程优化演进的一个方向。线程池就是一种对CPU利用的优化手段。

池化技术

池化技术就是提前保存大量的资源，以备不时之需。在机器资源有限的情况下，使用池化技术可以大大的提高资源的利用率，以此来提升性能。

在编程领域，一些常用的池化技术有(线程池、连接池、内在池、对象池等)。

带着问题看线程池

线程池的线程是如何做到复用的？

当线程小于核心线程数时，获取队列中的任务使用take方法，该方法中使用了await，这样一来，线程就不会被回收，而大于核心线程数以外的线程使用poll获取任务，当没有任务时poll超时会返回null，这时候该线程就会被退出。

我不太擅长文字描述，大家直接看下面的四张图，然后对照源码，我相信很容易就能看懂了。

线程池是如何做到高效并发的？

其实就是使用了Java的并发包中的AtomicInteger来记录状态以及线程数量，使用了ReentantLock，都是使用的CAS来进行处理，任何东西都不是绝对的，只有最适合，大家慢慢体会。

• 线程池状态和工作线程数量记录由AtomicInteger的value来记录，保证了原子性

• Worker中添加新Worker的时候，使用了ReentrantLock

• 工作线程Worker从等待队列中取任务的时候，这个由工作队列本身来保证线程安全，比如LinkedBlockingQueue等。

线程数量

• CPU密集型

  CPU使用率较高，也就是经常计算一些复杂的运算，逻辑处理等情况非常多的时候，线程数一般只需要设置为CPU核心数的线程个数就可以了。这一类型多出现在开发中的一些业务复杂计算和逻辑处理过程中。

• I/O密集型

  CPU使用率较低，程序中会存在大量的I/O操作占用时间，导致线程空余时间很多，所以通常就需要开CPU核心数两倍的线程。当线程进行I/O操作CPU空闲时，启用其他线程继续使用CPU，以提高CPU的使用率。线程等待时间所占比例越高，需要越多线程来继续使用CPU。这一类型在开发中主要出现在一些计算业务频繁的逻辑中。

一个计算为主的程序，多线程跑的时候，可以充分利用起所有的CPU核心数，但是如果线程数远超出CPU核心数量，反而会使得任务效率下降，因为频繁的切换线程也是要消耗时间的。因此对于CPU密集型的任务来说，线程数等于CPU数是最好的。

如果是一个磁盘或网络为主的程序，一个线程处于IO等待的时候，另一个线程还可以在CPU里面运行，有时候CPU闲着没事干，所有的线程都在等着IO，此时可以增加线程数，让CPU可以继续工作，但并不是线程越多越好，一般是CPU核心数的两倍是最佳的。

照图看源码

创建线程池

 package com.doaredo.test;  import org.junit.Test; import java.util.concurrent.*;  public class ThreadPoolDemo {  @Test public void threadPoolTest() { ExecutorService executorService = new ThreadPoolExecutor(1, 1, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10)); executorService.execute(new Runnable() { @Override public void run() { System.out.println("threadPool"); } }); executorService.shutdown(); } }

在了解线程池之前，我们需要先了解ThreadPoolExecutor构造函数中参数的意思。

• corePoolSize

  线程池的核心线程数，线程池没有任何任务的情况下，也会有核心线程池在等待任务的到来。

• maximumPoolSize

  最大线程数，不管提交了多少任务，线程池里最多工作线程数不会超过这个值。

• keepAliveTime

  线程的存活时间，当线程池里的线程数大于corePoolSize时，如果等待keepAliveTime时长还没有任务，则该线程退出。

• unit

  用来指定keepAliveTime的单位，比如秒：TimeUnit.SECONDS。

• workQueue

  阻塞队列，提交的任务将会被放到这个队列里

• threadFactory

  线程工厂，用来创建线程，主要是为了给线程起名字，默认工厂的线程名字：pool-1-thread-3。

• 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.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }

在读ThreadPoolExecutor源码之前，我们需要先了解以下知识点：

"|"或运算

"~"按位取反

"&"与运算(从高位开始比较，同为1则取1，否则取0)

ThreadPoolExecutor源码中定义了线程的五种状态：

1. RUNNING，运行状态，值是最小的，刚创建的线程池就是此状态。

2. SHUTDOWN，停工状态，不再接收新任务，已经接收的会继续执行。

3. STOP，停止状态，不再接收新任务，已经接收的正在执行的任务也会中断。

4. TIDYING，清空状态，所有任务都停止了，工作的线程也全部结束了。

5. TERMINATED，终止状态，线程池已销毁。

 //CAS private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); //线程池线程数的bit数(Integer.SIZE为32，这里就是29) private static final int COUNT_BITS = Integer.SIZE - 3; //最大线程数量(表达式结果为：536870911) //0001 1111 1111 1111 1111 1111 1111 1111(29位1) private static final int CAPACITY = (1 << COUNT_BITS) - 1;  // runState is stored in the high-order bits //1110 0000 0000 0000 0000 0000 0000 0000 private static final int RUNNING = -1 << COUNT_BITS; //0000 0000 0000 0000 0000 0000 0000 0000 private static final int SHUTDOWN = 0 << COUNT_BITS; //0010 0000 0000 0000 0000 0000 0000 0000 private static final int STOP = 1 << COUNT_BITS; //0100 0000 0000 0000 0000 0000 0000 0000 private static final int TIDYING = 2 << COUNT_BITS; //0110 0000 0000 0000 0000 0000 0000 0000 private static final int TERMINATED = 3 << COUNT_BITS;  // Packing and unpacking ctl private static int runStateOf(int c) { return c & ~CAPACITY; } private static int workerCountOf(int c) { return c & CAPACITY; } private static int ctlOf(int rs, int wc) { return rs | wc; }  /* * Bit field accessors that don't require unpacking ctl. * These depend on the bit layout and on workerCount being never negative. */  private static boolean runStateLessThan(int c, int s) { return c < s; }  private static boolean runStateAtLeast(int c, int s) { return c >= s; }  private static boolean isRunning(int c) { return c < SHUTDOWN; }

在ThreadPoolExecutor中使用AtomicInteger来记录线程的数量，但是只使用了AtomicInteger中value的低29位，高3位用来记录线程的状态，也就是上面的五种状态，分别由int类型的高3位来表示。

在java中一个int类型占4个字节，1个字节=8bit，也就是一个int类型有32bit。AtomicInteger中的value就是使用的int类型。

 private static int runStateOf(int c) { return c & ~CAPACITY; } private static int workerCountOf(int c) { return c & CAPACITY; }

我们看上面两个方法，初始计算的CAPACITY的二进制为29个1，高3位全为0，也就是0001 1111 1111 1111 1111 1111 1111 1111。

runStateOf(int c)、workerCountOf(int c)两个方法中的参数，在调用的时候传递的就是AtomicInteger的value值。这样在runStateOf方法中，通过value & CAPACITY(value与上CAPACITY的按位取反)，CAPACITY得到的全是0，然后跟value做与运算，最终得到的结果就是value的高3位结果，而这3位就表示的线程的状态。

同理，在workerCountOf方法，通过value & CAPACITY，最终得到的就是value的低29位，而这低29位就是表示的线程的数量。

execute/submit

线程池的工作就是以添加任务为驱动进行工作的。

线程池就是保存一系列线程的容器，在ThreadPoolExecutor中，将线程保存在一个HashSet中，为了防止线程处理不过来，还有一个用来保存任务的队列workQueue：

 private final HashSet<Worker> workers = new HashSet<Worker>(); private final BlockingQueue<Runnable> workQueue;

往线程池提交任务有两种方式，execute是ExecutorService接口定义的，submit有三种方法重载都在AbstractExecutorService(ExecutorService的实现)中定义，都是将要执行的任务包装为FutureTask来提交，使用者可以通过FutureTask来拿到任务的执行状态和执行最终结果，最终调用的都是execute方法。对于线程池来说，任务的状态是由FutureTask自己维护的，对线程池透明。

 public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; }  /** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ public <T> Future<T> submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task, result); execute(ftask); return ftask; }  /** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }

submit中都是调用的ThreadPoolExecutor中的execute方法：

 public void execute(Runnable command) { if (command == null) throw new NullPointerException(); //获取AtomicInteger中的value值 int c = ctl.get(); //判断当前线程数是否小于设置的核心线程数 if (workerCountOf(c) < corePoolSize) { //如果线程数没有达到设置的核心线程数，则直接添加一个核心线程 //提交任务后，会优先启动核心线程处理 if (addWorker(command, true)) return; //任务添加失败则重新获取value值 c = ctl.get(); } //如果线程池在运行状态，将任务添加到等待队列(如果队列已经满了就会返回false) if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); //如果线程池不是运行状态，将刚添加的任务从队列移除并且执行拒绝策略 if (! isRunning(recheck) && remove(command)) reject(command); //如果线程数为0，则添加一个非核心线程，并且不指定首次执行任务 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //添加非核心线程，指定首次执行任务，如果添加失败，执行异常策略 else if (!addWorker(command, false)) reject(command); }

addWorker

在分析addWorker方法前，先了解retry这个标记，retry不是关键字，它是一个标记，标记后面一般跟循环，功能类似于goto，直接上案例：

 package com.doaredo.test;  import org.junit.Test;  public class RetryDemo {  @Test public void retryContinue(){ retry: for (int i = 0; i < 2; i++){ for (int j =0; j < 5; j++){ System.out.print(j); if(j == 3){ continue retry; } } } System.out.println("====================="); }  @Test public void retryBreak(){ retry: for (int i = 0; i < 2; i++){ for (int j =0; j < 5; j++){ System.out.print(j); if(j == 3){ break retry; } } } System.out.println("====================="); } }

retryContinue()输出结果：01230123=====

retryBreak()输出结果：0123=====

我相信看了案例，大家应该就明白了，continue retry;就是跳转到外层循环继续执行，而break retry;则是直接跳出整个循环。

addWorker方法分析

先看retry后面的双重for循环，第9-13行代码应该是这里比较难理解的，我们后面再来分析，这里我们直接无视这部分代码。2-29行代码，主要做的功能就是22行代码中将线程数+1。

 private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); //如果线程池的状态在SHUTDOWN或之上的状态，只有一种情况还需要继续添加线程 //这种情况就是：线程池已经SHUTDOWN，但是队列中还有任务在排队，而且不接受新任务 //这里还继续添加线程是为了加快执行等待队列中的任务，尽快让线程池关闭 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false;  for (;;) { int wc = workerCountOf(c); //如果线程数超过理论最大容量，如果core是true跟最大核心线程数比较，否则跟最大线程数比较 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //CAS操作，增加线程数+1，成功则跳出双层循环 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } //到这里说明已经成功的将线程数+1，但是真正的线程还没有被添加 boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //Worker继承了AQS，实现了Runnable接口的包装类  w = new Worker(firstTask); 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()); //检查线程状态，跟前面一样，只有当线程池处于RUNNING或者处于SHUTDOWN并且firstTask==null的时候，这时候创建Worker来加速处理队列中的任务 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { //线程只能被start一次 if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }

execute方法没有加锁，但是在addWorker方法内部加锁了，可以保证不会创建超过预期的线程数。

worker是怎么工作的

worker的run方法调用的是ThreadPoolExecutor的runWorker方法。

 final void runWorker(Worker w) { Thread wt = Thread.currentThread(); //取出需要执行的任务 Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //task不为null，或者队列中能取到任务(这里getTask()会阻塞) while (task != null || (task = getTask()) != null) { //这个lock在这里是为了如果线程被中断，那么会抛出InterruptedException，而退出循环，结束线程 w.lock(); // 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 { //任务开始执行前的钩子方法 beforeExecute(wt, task); Throwable thrown = null; try { 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; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { //退出worker processWorkerExit(w, completedAbruptly); } } private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out?  for (;;) { int c = ctl.get(); int rs = runStateOf(c);  // Check if queue empty only if necessary. //检查线程池的状态，如果已经是STOP及以上的状态，或者已经SHUTDOWN，队列也是空的时候，直接return null，并将Worker数量-1 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; }  int wc = workerCountOf(c);  // Are workers subject to culling? // 注意这里的allowCoreThreadTimeOut参数，字面意思是否允许核心线程超时，即如果我们设置为false，那么只有当线程数wc大于corePoolSize的时候才会超时 //更直接的意思就是，如果设置allowCoreThreadTimeOut为false，那么线程池在达到corePoolSize个工作线程之前，不会让闲置的工作线程退出 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //确认超时，将Worker数-1，然后返回 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; }  try { //从队列中取任务，根据timed选择是有时间期限的等待还是无时间期限的等待 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }

线程池优点

线程池复用线程有以下优点：

• 减少资源创建：减少内存开销，创建线程占用内存

• 降低系统开销：创建线程需要时间，会延迟处理的请求

• 提高稳定性：避免无限创建线程引起内存溢出

Executors

在阿里开发手册上有一条：线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式来创建，具体是什么原因往后看。

Executors创建线程池的方式

• 创建返回ThreadPoolExecutor对象

• 创建返回ScheduleThreadPoolExecutor对象

• 创建返回ForkJoinPool对象

前面我们讲了ThreadPoolExecutor的源码，我们以创建返回ThreadPoolExecutor对象的方式为例，返回ThreadPoolExecutor对象共有三种方法：

• newCachedThreadPool(创建可缓存的线程池)

• newSingleThreadExecutor(创建单线程的线程池)

• newFixedThreadPool(创建固定长度的线程池)

Executors源码如下：

 /* 和newSingleThreadExecutor类似 */ public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } /* 当一个任务提交时，首先会创建一个核心线程来执行任务，如果超过核心线程的数量，将会放入队列中 因为LinkedBlockingQueue的长度为Integer.MAX_VALUE，因此队列中中以插入非常多的任务，从而引起OOM 因为队列是无界队列，maximumPoolSize和keepAliveTime参数将无效，这样就不会创建非核心线程 */ public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); } /* 当一个任务提交时，corePoolSize为0不创建核心线程 SynchronousQueue是一个不存储元素的队列，因此最终会创建非核心线程来执行任务 非核心线程空闲60s将被回收，而Integer.MAX_VALUE最大线程数非常大，这样基本可以无限创建线程 在资源有限的情况下容易引起OOM异常 */ public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }

OOM异常测试

 @Test public void oomTest(){ ExecutorService executorService = Executors.newCachedThreadPool(); AtomicInteger i = new AtomicInteger(0); for(;;){ executorService.submit(()->{ System.out.println(i.getAndIncrement()); }); } }

设置VM options：-Xms5M -Xmx5M

运行结果会抛出异常：

 Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "pool-1-thread-5154"