深入理解线程池

一、线程池

1.1、线程池的优势

降低资源消耗。

通过重复利用已创建的线程降低线程创建和销毁造成的损耗

提高响应速度

当任务到达时，任务可以不需要等到线程创建就能立即执行

提高线程的可管理性

线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

线程池一般用于执行多个不相关联的耗时任务，没有多线程的情况下，任务顺序执行，使用了线程池的话可以让多个不相关联的任务同时执行

1.2、Executor框架介绍

Executor框架是Java5之后引进的，在Java5之后，通过Executor来启动线程比使用Thread的start方法更好，除了是线程池更易管理、效率更好以外，还可以避免this逃逸问题

this逃逸是指，在构造函数返回之前其他线程就持有该对象的引用，调用尚未构造完全的对象的方法可能引发令人疑惑的错误

在这里插入图片描述

线程池的真正实现类是ThreadPoolExecutor，它一共有四种构造方法如下：

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对象，需要如下几个参数：

corePoolSize(必需):核心线程数，默认情况下，核心线程数会一直存活，但是当 allowCoreThreadTimeout设置为true时，核心线程也会超时回收
maximumPoolSize（必需）：线程池所能容纳的最大线程数量。当活跃线程数到达该数值后，后续的新任务将会阻塞
keepAliveTime（必需）：线程闲置超时时长。如果超过该时长，非核心线程就会被回收 。（当 allowCoreThreadTimeout设置为true时，核心线程也会超时回收）
unit（必需）：指定keepAliveTime参数的时间单位。常用的有：TimeUnit.MILLISECONDS（毫秒）、TimeUnit.SECONDS（秒）、TimeUnit.MINUTES（分）
workQueue（必需）：任务队列。通过线程池的execute（）方法提交的Runnable对象将存储在该参数中。其采用阻塞队列实现
threadFactory（可选）：线程工厂，用于指定为线程池创建线程的方式
handler（可选）：拒绝策略。当达到最大线程时需要执行的饱和策略

3.3、线程池的工作原理

在这里插入图片描述

写一个实战demo

MyRunnable类，实现Runnable类

/**
 * 这是一个简单的Runnable类，需要大约5秒钟来执行其任务。
 */
public class MyRunnable implements Runnable {
    private String command;

    public MyRunnable(String s) {
        this.command = s;
    }


    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " Start. Time = " + new Date());
        processCommand();
        System.out.println(Thread.currentThread().getName() + " End. Time = " + new Date());
    }

    private void processCommand() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        return this.command;
    }
}

线程池创建线程

public class ThreadPoolDemo {
    private static final int CORE_POOL_SIZE = 5;
    private static final int MAX_POOL_SIZE = 10;
    private static final int QUEUE_CAPACITY = 100;
    private static final Long KEEP_ALIVE_TIME = 1L;
    public static void main(String[] args) {

        //使用阿里巴巴推荐的创建线程池的方式
        //通过ThreadPoolExecutor构造函数自定义参数创建
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy());

        for (int i = 0; i < 10; i++) {
            //创建WorkerThread对象（WorkerThread类实现了Runnable 接口）
            Runnable worker = new MyRunnable("" + i);
            //执行Runnable
            executor.execute(worker);
        }
        //终止线程池
        executor.shutdown();
        while (!executor.isTerminated()) {//当executor.isTerminated()判断为true，跳出循环。当且仅当调用了shutdown方法后并且所有提交的任务都完成后返回true
        }
        System.out.println("Finished all threads");
    }
}

代码输出结果可以看出：线程池会先执行5个任务，然后这些任务中有执行完的话，就会去拿新的任务执行。

executor.execute(worker);

executor方法很重要，源码如下

   // 存放线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)
   private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    private static int workerCountOf(int c) {
        return c & CAPACITY;
    }
    //任务队列
    private final BlockingQueue<Runnable> workQueue;

    public void execute(Runnable command) {
        // 如果任务为null，则抛出异常。
        if (command == null)
            throw new NullPointerException();
        // ctl 中保存的线程池当前的一些状态信息
        int c = ctl.get();

        //  下面会涉及到 3 步 操作
        // 1.首先判断当前线程池中执行的任务数量是否小于 corePoolSize
        // 如果小于的话，通过addWorker(command, true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 2.如果当前执行的任务数量大于等于 corePoolSize 的时候就会走到这里，表明创建新的线程失败。
        // 通过 isRunning 方法判断线程池状态，线程池处于 RUNNING 状态并且队列可以加入任务，该任务才会被加入进去
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            // 再次获取线程池状态，如果线程池状态不是 RUNNING 状态就需要从任务队列中移除任务，并尝试判断线程是否全部执行完毕。同时执行拒绝策略。
            if (!isRunning(recheck) && remove(command))
                reject(command);
                // 如果当前工作线程数量为0，新创建一个线程并执行。
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //3. 通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
        // 传入 false 代表增加线程时判断当前线程数是否少于 maxPoolSize
        //如果addWorker(command, false)执行失败，则通过reject()执行相应的拒绝策略的内容。
        else if (!addWorker(command, false))
            reject(command);
    }

简单来说，executor.execute(worker)，执行这行代码，将一个任务提交到线程池中，步骤如下：

如果当前线程数小于核心线程数，那么新建一个线程来执行任务
如果当前线程数大于或等于核心线程数，（小于最大线程数）但是任务队列还没满，那么将该任务放到任务队列中
如果当前任务队列已经满了，但是线程数小于最大线程数，就创建一个线程来执行任务
如果当前运行的线程数大于等于最大线程数，那么当前任务会被拒绝，通过reject（）执行相应的拒绝策略的内容

多次调用到addWorker 方法，这个方法主要用来创建新的工作线程，如果返回true说明创建和启动工作线程成功，否则的话返回的就是false

这其实是线程复用 ，线程池使用固定数量或可变数量的线程去循环执行任务，将任务和线程分开来，任务有一个任务池（任务队列），线程数量则由我们自己定义。当线程手上的任务完成，就循环不断的从任务池里拿任务执行（调用run方法），这就使得创建一定数量的线程来复用不断执行任务。

3.4、线程池的参数

1、任务队列（workQueue）

任务队列是基于阻塞队列实现的，即基于生产者消费者模式，线程安全的高效队列。

Java为我们提供了7种阻塞队列的实现

不同的线程池会选用不同的阻塞队列，我们可以结合内置线程池来分析。

容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue（无界队列）：FixedThreadPool 和 SingleThreadExector 。由于队列永远不会被放满，因此FixedThreadPool最多只能创建核心线程数的线程。
SynchronousQueue（同步队列）：CachedThreadPool 。SynchronousQueue 没有容量，不存储元素，目的是保证对于提交的任务，如果有空闲线程，则使用空闲线程来处理；否则新建一个线程来处理任务。也就是说，CachedThreadPool 的最大线程数是 Integer.MAX_VALUE ，可以理解为线程数是可以无限扩展的，可能会创建大量线程，从而导致 OOM。
DelayedWorkQueue（延迟阻塞队列）：ScheduledThreadPool 和 SingleThreadScheduledExecutor 。DelayedWorkQueue 的内部元素并不是按照放入的时间排序，而是会按照延迟的时间长短对任务进行排序，内部采用的是“堆”的数据结构，可以保证每次出队的任务都是当前队列中执行时间最靠前的。DelayedWorkQueue 添加元素满了之后会自动扩容原来容量的 1/2，即永远不会阻塞，最大扩容可达 Integer.MAX_VALUE，所以最多只能创建核心线程数的线程。

2、线程工厂（threadFactory）

线程工厂用来指定创建线程的方式。该参数可以不用指定，Executors框架为我们实现了一个默认的线程工厂

/**
 * The default thread factory.
 */
private static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
 
    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }
 
    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

3、拒绝策略（handler）

当线程池的线程数达到最大线程数时，需要执行拒绝策略。拒绝策略需要实现RejectExecutionHandler接口。Executors框架为我们实现了四种拒绝策略

AbortPolicy（默认）：丢弃任务并抛出 RejectedExecutionException 异常。
CallerRunsPolicy：由调用线程处理该任务。
DiscardPolicy：丢弃任务，但是不抛出异常。可以配合这种模式进行自定义的处理方式。
DiscardOldestPolicy：丢弃队列最早的未处理任务，然后重新尝试执行任务。

  ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(QUEUE_CAPACITY),
                //拒绝策略
                new ThreadPoolExecutor.CallerRunsPolicy());

3.5、几种常见的内置线程池

Executors已经为我们封装好了4中常见的功能线程池。

1、定长线程池（FixedThreadPool）相关源码

   /**
     * 创建一个可重用固定数量线程的线程池
     */
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

实现原理：

在FixedThreadPool中，corePoolSize和maximumPoolSizes都被设置成nThreads，这个nThreads是我们使用时自己传递的。即使最大线程数量比核心线程数量大，也最多只会创建核心线程数量的线程，因为FixedThreadPool使用的队列是容量为Integer.MAX_VALUE的LinkedBlockingQueue（无界队列）。队列用于不会放满，所以和最大线程数量无关。

当新任务来时，如果当前运行的线程数小于corePoolSize，创建新的线程来执行任务；如果当前运行的线程数大于等于corePoolSize，就将任务加入到任务队列，等核心线程执行完任务会在循环中反复从队列中获取任务来执行。

为什么不推荐使用FixedThreadPool

FixedThreadPool使用的无界队列LinkedBlockingQueue作为线程池的任务队列，会有以下影响：

1）线程池的线程数达到corePoolSize后，新任务将在无界队列中等待，因此线程池中的线程数不会超过corePoolSize

2）使用无界队列时，maximumPoolSize将是一个无效参数，因为不可能存在任务队列满的情况

3）同样，使用无界队列时，keepAliveTime也是一个无效参数

4)运行中的FixedThreadPool不会拒绝任务，在任务比较多的时候会导致OOM（内存溢出）。—一直堆在阻塞队列中

应用场景：控制线程最大并发数

// 1. 创建定长线程池对象 & 设置线程池线程数量固定为3
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
// 2. 创建好Runnable类线程对象 & 需执行的任务
Runnable task =new Runnable(){
  public void run() {
     System.out.println("执行任务啦");
  }
};
// 3. 向线程池提交任务
fixedThreadPool.execute(task);

2、定时线程池（ScheduledThreadPool）

核心线程数量固定，非核心线程数量无限，执行完闲置10ms后回收，任务队列为延时阻塞队列
应用场景：执行定时或周期性的任务

// 1. 创建 定时线程池对象 & 设置线程池线程数量固定为5
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
// 2. 创建好Runnable类线程对象 & 需执行的任务
Runnable task =new Runnable(){
  public void run() {
     System.out.println("执行任务啦");
  }
};
// 3. 向线程池提交任务
scheduledThreadPool.schedule(task, 1, TimeUnit.SECONDS); // 延迟1s后执行任务
scheduledThreadPool.scheduleAtFixedRate(task,10,1000,TimeUnit.MILLISECONDS);// 延迟10ms后、每隔1000ms执行任务

3、可缓存线程池（CachedThreadPool）

无核心线程，最大线程数量为Integer.MAX.VALUE，即无界，执行完闲置60s后回收，任务队列为不存储元素的阻塞队列
应用场景：执行大量、耗时少的任务

这也就意味着如果主线程提交任务的速度高于 maximumPool中线程处理任务的速度时，CachedThreadPool 会不断创建新的线程。极端情况下，这样会导致耗尽 cpu 和内存资源。

// 1. 创建可缓存线程池对象
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
// 2. 创建好Runnable类线程对象 & 需执行的任务
Runnable task =new Runnable(){
  public void run() {
     System.out.println("执行任务啦");
  }
};
// 3. 向线程池提交任务
cachedThreadPool.execute(task);

缺陷：会创建大量线程，从而导致OOM

4、单线程化线程池（SingleThreadExecutor）

corePoolSize和maximumPoolSize都是1，队列和FixedThreadPool一样是无边界的链表队列
原理也和FixedThreadPool一样，不过是现在的核心线程数设置为1
应用场景：不适合并发

// 1. 创建单线程化线程池
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
// 2. 创建好Runnable类线程对象 & 需执行的任务
Runnable task =new Runnable(){
  public void run() {
     System.out.println("执行任务啦");
  }
};
// 3. 向线程池提交任务
singleThreadExecutor.execute(task);

缺陷也很明显。因为是采用无边界的链表阻塞队列，线程数达到核心线程数1后，再来的新任务就会一直往阻塞队列里堆，导致OOM！

5、不推荐使用Executors来创建线程池

阿里巴巴Java开发手册中强制线程池不允许使用Executors去创建，而是要通过ThreadPoolExecutor构造方法来创建。这样的创建方式可以让我们更好的理解线程池的运行规则，规避资源耗尽的风险。

3.6、几个常见对比

isTerminated（）和 isShutdown（）

isShutdown 当调用shutdown（）方法后返回true
isTerminated 当调用shutdown（）方法后，并且所有提交的的任务完成后返回true

shutdown()和shutdownNow()

shutdown（） :关闭线程池，线程池的状态变为 SHUTDOWN。线程池不再接受新任务了，但是队列里的任务得执行完毕。
shutdownNow（） :关闭线程池，线程池的状态变为STOP。线程池会终止当前正在运行的任务，并停止处理排队的任务并返回正在等待执行的 List

一般会用shutdown

execute() 和 submit()

execute()方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功与否；
submit()方法用于提交需要返回值的任务。线程池会返回一个 Future 类型的对象，通过这个 Future对象可以判断任务是否执行成功， 并且可以通过 Future 的 get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用 get（long timeout，TimeUnit unit）方法的话，如果在 timeout 时间内任务还没有执行完，就会抛出 java.util.concurrent.TimeoutException

ExecutorService executorService = Executors.newFixedThreadPool(3);

Future<String> submit = executorService.submit(() -> {
    try {
        Thread.sleep(5000L);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "abc";
});

String s = submit.get();
System.out.println(s);
executorService.shutdown();

3.7、最佳实践

1、正确声明线程池

线程池必须通过ThreadPoolExecutor的构造函数来声明，避免使用Executors类创建线程池，会有OOM风险

2、建议不同类别的业务用不同的线程池

一般建议是不同的业务使用不同的线程池，配置线程池的时候根据当前业务的情况对当前线程池进行配置，因为不同的业务的并发以及对资源的使用情况都不同，重心优化系统性能瓶颈相关的业务。

3、给线程池命名

初始化线程池的时候需要显示命名（设置线程池名称前缀），有利于定位问题

默认情况下，创建的线程名字类似于 pool-1-thread-n这样的，没有业务含义，不利于定位问题。

给线程池里的线程命名通常有一下两种方式（创建线程工厂时命名）

利用guava的ThreadFactoryBuilder

ThreadFactory threadFactory = new ThreadFactoryBuilder()
                        .setNameFormat(threadNamePrefix + "-%d")
                        .setDaemon(true).build();
ExecutorService threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory)

自己实现线程工厂

import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
/**
 * 线程工厂，它设置线程名称，有利于我们定位问题。
 */
public final class NamingThreadFactory implements ThreadFactory {

    private final AtomicInteger threadNum = new AtomicInteger();
    private final ThreadFactory delegate;
    private final String name;

    /**
     * 创建一个带名字的线程池生产工厂
     */
    public NamingThreadFactory(ThreadFactory delegate, String name) {
        this.delegate = delegate;
        this.name = name; // TODO consider uniquifying this
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = delegate.newThread(r);
        t.setName(name + " [#" + threadNum.incrementAndGet() + "]");
        return t;
    }

}

4、正确配置线程池参数

线程池的大小不应该过大也不应该过小

如果我们设置的线程池数量太小的话，如果同一时间有大量任务/请求需要处理，可能会导致大量的请求/任务在任务队列中排队等待执行，甚至会出现任务队列满了之后任务/请求无法处理的情况，或者大量任务堆积在任务队列导致OOM。这样很明显是有问题的，CPU根本没有得到充分利用
如果我们设置的线程数量太大，大量线程可能会同时在争取CPU资源，这样会导致大量的上下文切换，从而增加线程的执行时间，影响了整体执行效率。

上下文切换：多线程编程中一般线程的个数都大于CPU核心的个数，而一个CPU核心在任意时刻只能被一个线程使用，为了让这些线程都得到有效执行，CPU采取的策略是为每个线程分配时间片并轮转的形式，当一个线程的时间片用完的时候就会重新处于就绪状态让给其他线程使用 ，这个过程就属于一次上下文切换。

有一个简单并且适用面比较广的公式：

对于CPU密集型任务（N+1）：这种任务消耗的主要是CPU资源，可以将线程数设置为N（CPU核心数）+1。比CPU核心数多出来的一个线程是为了防止线程偶发的缺页中断，或者其他原因导致的任务暂停而带来的影响。一旦任务暂停，CPU就会处于空闲状态，这种情况下多的一个线程就可以充分利用CPU的空闲时间
对于I/O密集型任务（2N）：这种任务应用起来，系统会用大部分的时间来处理I/O交互，而线程在处理I/O的时间段内不会占用CPU来处理，这时可以将CPU交给其他线程使用。因此在I/O密集型任务的应用中，我们可以多配置一些线程，具体的计算方法是2N

CPU密集型简单理解就是利用CPU计算能力的任务，比如你在内存中对大量数据进行排序。

但凡涉及到网络读取，文件读取这类都是IO密集型，这类任务的特点是CPU计算耗费时间相当于等待IO操作完成的时间来说很少，大部分时间都花在IO操作上（所以可以分出cpu去处理线程）

5、别忘记关闭线程池

当线程池不再需要使用时，应该显式的关闭线程池，释放线程资源。

线程池提供了两个关闭线程池的方法，shutdown（）和shutdownNow（）。调用完关闭线程的方法后，并不代表线程池已经完成关闭操作，它只是异步的通知线程池要关闭了！如果要同步等待线程池彻底关闭才继续往下执行，需要调用awaitTermination方法进行同步等待

// ...
// 关闭线程池
executor.shutdown();
try {
    // 等待线程池关闭，最多等待5分钟
    if (!executor.awaitTermination(5, TimeUnit.MINUTES)) {
        // 如果等待超时，则打印日志
        System.err.println("线程池未能在5分钟内完全关闭");
    }
} catch (InterruptedException e) {
    // 异常处理
}