Java面试篇（线程池相关专题）

1. 为什么要使用线程池

我们为什么要使用线程池呢，主要有两个原因：

1. 每次创建线程的时候都会占用一定的内存空间，如果要创建很多个线程，有可能会浪费内存，严重的情况下还有可能会导致内存溢出
2. CPU 资源是有限的，同一时刻 CPU 只能处理一个线程，如果有大量的请求到达服务器，我们创建了大量的线程，那么很多线程都没有 CPU 的执行权，这些线程都需要等待获取 CPU 的执行权，在这个过程中会有非常多的切换线程操作，频繁的线程切换操作会导致上下文切换开销增大，CPU 花费在真正执行任务上的时间就会减少，从而导致程序的性能下降

在项目开发的过程中，一般都会用线程池来管理线程、创建线程

线程池相关的内容，跟实际开发是有很大关系的，面试官也是特别喜欢问

2. 线程池的核心参数和线程池的执行原理

2.1 线程池的核心参数

线程池的核心参数主要有七个，我们主要参考 ThreadPoolExecutor 类的具有七个参数的构造函数，这七个参数也是面试官提问的重点

1. corePoolSize：核心线程数
2. maximumPoolSize：最大线程数（核心线程数 + 救急线程数的最大值）
3. keepAliveTime：救急线程的生存时间，如果生存时间内没有新任务，与救急线程相关的资源会被释放
4. unit：救急线程的生存时间单位
5. workQueue：当没有空闲的核心线程时，新来任务会加入到此队列中排队，如果队列满了会创建救急线程来执行任务
6. threadFactory：线程工厂，可以定制如何线程对象，例如为线程设置名字、是否为守护线程等
7. handler：拒绝策略，当所有线程都在繁忙、workQueue 中的线程也满了时，会触发拒绝策略

2.2 线程池的执行原理

下面为大家介绍一下线程池的执行原理，也就是线程池是怎么工作的

首先，当有一个新任务到来时，会先判断核心线程数是否已经满了，如果核心线程数没满，就将任务添加到工作线程中，执行这个任务

如果核心线程数已经满了，会判断阻塞队列是否满了，如果阻塞队列中还有空间，就把任务添加到阻塞队列中进行等待，

如果阻塞队列满了，会判断线程数是否小于最大线程数，如果线程数小于最大线程数，会创建救急线程来执行任务

当核心线程或救急线程处于空闲的时候，会去阻塞队列中检查一下是否有需要执行的任务，如果有，就会使用核心线程或救急线程来执行阻塞队列中的任务

假如所有条件都不满足，线程池会有一个拒绝策略，拒绝策略有以下四种：

1. AbortPolicy：直接抛出异常，默认使用的拒绝策略
2. CallerRunsPolicy：用调用者所在的线程来执行任务
3. DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务
4. DiscardPolicy：直接丢弃任务

可以运行以下代码，查看控制台的输出，辅助理解线程池的执行原理

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class TestThreadPoolExecutor {

    static class MyTask implements Runnable {

        private final String name;

        private final long duration;

        public MyTask(String name) {
            this(name, 0);
        }

        public MyTask(String name, long duration) {
            this.name = name;
            this.duration = duration;
        }

        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getName() + " Running..." + this);
                Thread.sleep(duration);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
        }

        @Override
        public String toString() {
            return "MyTask(" + name + ")";
        }
    }

    public static void main(String[] args) {
        AtomicInteger atomicInteger = new AtomicInteger(1);
        ArrayBlockingQueue<Runnable> arrayBlockingQueue = new ArrayBlockingQueue<>(2);

        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
                2,
                3,
                0,
                TimeUnit.MILLISECONDS,
                arrayBlockingQueue,
                runnable -> new Thread(runnable, "myThread" + atomicInteger.getAndIncrement()),
                new ThreadPoolExecutor.AbortPolicy());
        // new ThreadPoolExecutor.CallerRunsPolicy());
        // new ThreadPoolExecutor.DiscardOldestPolicy());
        // new ThreadPoolExecutor.DiscardPolicy());
        showState(arrayBlockingQueue, threadPool);
        threadPool.submit(new MyTask("1", 3600000));
        showState(arrayBlockingQueue, threadPool);
        threadPool.submit(new MyTask("2", 3600000));
        showState(arrayBlockingQueue, threadPool);
        threadPool.submit(new MyTask("3"));
        showState(arrayBlockingQueue, threadPool);
        threadPool.submit(new MyTask("4"));
        showState(arrayBlockingQueue, threadPool);
        threadPool.submit(new MyTask("5", 3600000));
        showState(arrayBlockingQueue, threadPool);
        threadPool.submit(new MyTask("6"));
        showState(arrayBlockingQueue, threadPool);
    }

    private static void showState(ArrayBlockingQueue<Runnable> queue, ThreadPoolExecutor threadPool) {
        try {
            Thread.sleep(300);
        } catch (InterruptedException exception) {
            exception.printStackTrace();
        }
        List<Object> tasks = new ArrayList<>();
        for (Runnable runnable : queue) {
            try {
                Field callable = FutureTask.class.getDeclaredField("callable");
                callable.setAccessible(true);
                Object adapter = callable.get(runnable);
                Class<?> clazz = Class.forName("java.util.concurrent.Executors$RunnableAdapter");
                Field task = clazz.getDeclaredField("task");
                task.setAccessible(true);
                Object object = task.get(adapter);
                tasks.add(object);
            } catch (Exception exception) {
                exception.printStackTrace();
            }
        }
        System.err.println("pool size: " + threadPool.getPoolSize() + ", queue: " + tasks);
    }

}

如果运行代码时遇到了以下错误，运行前可以添加以下 VM 参数

错误信息：

java.lang.reflect.InaccessibleObjectException: Unable to make field private java.util.concurrent.Callable java.util.concurrent.FutureTask.callable accessible: module java.base does not "opens java.util.concurrent" to unnamed module @4fca772d
	at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:354)
	at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:297)
	at java.base/java.lang.reflect.Field.checkCanSetAccessible(Field.java:178)
	at java.base/java.lang.reflect.Field.setAccessible(Field.java:172)

VM 参数的具体内容：

--add-opens java.base/java.util.concurrent=ALL-UNNAMED 

3. 线程池中常见的阻塞队列

workQueue，当没有空闲的核心线程时，新来任务会加入到阻塞队列，阻塞队列满了后会创建救急线程执行任务

3.1 常见的阻塞队列

常见的阻塞队列主要有以下四种（重点了解 ArrayBlockingQueue 和 LinkedBlockingQueue）：

1. ArrayBlockingQueue：基于数组结构的有界阻塞队列，符合队列的先进先出原则
2. LinkedBlockingQueue：基于链表结构的有界阻塞队列，符合队列的先进先出原则
3. DelayedWorkQueue：优先级队列，能够保证每次出队的元素是队列中延迟时间最短的任务
4. SynchronousQueue：不存储元素的阻塞队列，每次执行插入操作前都必须等待一个移出操作

补充：与 DelayedWorkQueue 相关的知识

任何想要放入 DelayedWorkQueue 的对象都必须实现 Delayed 接口，Delayed 接口要求实现两个方法：

1. getDelay(TimeUnit unit) 方法用于返回元素的剩余延迟时间，即距离执行时间还有多长时间
2. compareTo(Delayed other) 方法用于比较两个 Delayed 对象的延迟时间

3.2 ArrayBlockingQueue 和 LinkedBlockingQueue 的区别

4. 如何确定线程池的核心线程数

4.1 应用程序中任务的类型

应用程序中任务的类型主要可以分为两种：

1. IO 密集型任务
2. CPU 密集型任务

4.1.1 IO 密集型任务

常见的 IO 密集型任务主要有文件读写、数据库读写、网络请求等

4.1.2 CPU 密集型任务

常见的 CPU 密集型任务主要有计算较为密集的代码、BitMap 转换、JSON 转换等

4.2 如何确定核心线程数

对于 IO 密集型的任务来说，核心线程数可以设置为 2 * N + 1 （其中 N 为 CPU 的核数），因为 IO 密集型任务消耗的 CPU 资源较少

对于 CPU 密集型的任务来说，核心线程数可以设置为 N + 1 （其中 N 为 CPU 的核数），因为 CPU 密集型任务需要消耗大量的 CPU 资源，线程数少了，就能够减少 CPU 在不同线程之间切换所耗费的时间，充分地利用 CPU 资源

一般来说，用 Java 开发的应用程序，任务的类型大都为 IO 密集型

如何查看电脑的 CPU 核数呢，可以运行以下代码查看

public class ProcessorsDemo {

    public static void main(String[] args) {
        System.out.println(Runtime.getRuntime().availableProcessors());
    }

}

5. 线程池的种类

在 java.util.concurrent.Executors 类中提供了大量创建连接池的静态方法，常见的有四种：

1. 创建使用固定线程数的线程池
2. 创建单线程线程池
3. 创建可缓存线程池
4. 创建提供延迟功能和周期执行功能的线程池

5.1 固定线程数的线程池

我们可以查看创建固定线程数线程池的源码

固定线程数的线程池的特点是：

1. 核心线程数与最大线程数一样，没有救急线程
2. 阻塞队列是 LinkedBlockingQueue ，最大容量为 Integer.MAX_VALUE

固定线程数的线程池适用于任务量已知、耗时较长的任务

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FixedThreadPoolCase {

    static class FixedThreadDemo implements Runnable {
        @Override
        public void run() {
            String threadName = Thread.currentThread().getName();
            for (int i = 0; i < 2; i++) {
                System.out.println(threadName + ":" + i);
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 创建一个固定大小的线程池，核心线程数和最大线程数都是3
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        for (int i = 0; i < 5; i++) {
            executorService.submit(new FixedThreadDemo());
            Thread.sleep(10);
        }

        executorService.shutdown();
    }

}

5.2 单线程线程池

单线程线程池只会用唯一的工作线程来执行任务，保证所有任务按照先进先出的顺序（FIFO）执行

我们可以查看创建单线程线程池的源码

单线程线程池的特点是：

1. 核心线程数和最大线程数都是1
2. 阻塞队列是LinkedBlockingQueue ，最大容量为Integer.MAX VALUE

单线程线程池适用于需要严格按照顺序执行的任务

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SingleThreadCase {

    static int count = 0;

    static class Demo implements Runnable {
        @Override
        public void run() {
            count++;
            System.out.println(Thread.currentThread().getName() + ":" + count);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 单个线程池，核心线程数和最大线程数都是1
        ExecutorService executorService = Executors.newSingleThreadExecutor();

        for (int i = 0; i < 10; i++) {
            executorService.execute(new Demo());
            Thread.sleep(5);
        }
        executorService.shutdown();
    }

}

5.3 可缓存线程池

我们可以查看创建可缓存线程池的源码

可缓存线程池的特点：

1. 核心线程数为 0
2. 最大线程数是 Integer.MAX VALUE
3. 阻塞队列为 SynchronousQueue：不存储元素的阻塞队列，每个插入操作都必须等待一个移出操作

可缓存线程池适用于任务数比较密集，但每个任务执行时间较短的情况

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CachedThreadPoolCase {

    static class Demo implements Runnable {
        @Override
        public void run() {
            String threadName = Thread.currentThread().getName();
            try {
                // 修改睡眠时间，模拟线程执行需要花费的时间
                Thread.sleep(100);

                System.out.println(threadName + "执行完了");
            } catch (InterruptedException interruptedException) {
                interruptedException.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 创建一个缓存的线程，没有核心线程数，最大线程数为Integer.MAX_VALUE
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            executorService.execute(new Demo());
            Thread.sleep(1);
        }
        executorService.shutdown();
    }

}

5.4 提供延迟功能和周期执行功能的线程池

我们可以查看创建提供延迟功能和周期执行功能的线程池的源码

顾名思义，提供延迟功能和周期执行功能的线程池适用于需要延迟执行或周期执行的任务

import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledThreadPoolCase {

    static class Task implements Runnable {
        @Override
        public void run() {
            try {
                String threadName = Thread.currentThread().getName();

                System.out.println(threadName + ", 开始：" + new Date());
                Thread.sleep(1000);
                System.out.println(threadName + ", 结束：" + new Date());

            } catch (InterruptedException interruptedException) {
                interruptedException.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 按照周期执行的线程池，核心线程数为2，最大线程数为Integer.MAX_VALUE
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2);
        System.out.println("程序开始：" + new Date());

        /*
         * schedule方法：提交任务到线程池中
         * 第一个参数：提交的任务
         * 第二个参数：任务执行的延迟时间
         * 第三个参数：时间单位
         */
        scheduledThreadPool.schedule(new Task(), 0, TimeUnit.SECONDS);
        scheduledThreadPool.schedule(new Task(), 1, TimeUnit.SECONDS);
        scheduledThreadPool.schedule(new Task(), 5, TimeUnit.SECONDS);

        Thread.sleep(5000);

        // 关闭线程池
        scheduledThreadPool.shutdown();
    }

}

6. 为什么不建议使用 Executors 类提供的静态方法创建线程池

阿里开发手册《Java开发手册-嵩山版》中指出

7. 线程池（多线程）的使用场景

7.1 CountDownLatch

CountDownLatch：闭锁、倒计时锁，用来进行线程同步协作，等待所有线程完成倒计时（一个或者多个线程，等待其他多个线程完成某件事情之后才能执行）

1. 构造参数用来初始化等待计数值
2. await()方法用来等待计数归零
3. countDown()方法用来让计数减一

import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {

    public static void main(String[] args) throws InterruptedException {
        // 初始化了一个参数为 3 的倒计时锁
        CountDownLatch countDownLatch = new CountDownLatch(3);

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "-begin...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            countDownLatch.countDown();
            System.out.println(Thread.currentThread().getName() + "-end..." + countDownLatch.getCount());
        }).start();

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "-begin...");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            countDownLatch.countDown();
            System.out.println(Thread.currentThread().getName() + "-end..." + countDownLatch.getCount());
        }).start();

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "-begin...");
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            countDownLatch.countDown();
            System.out.println(Thread.currentThread().getName() + "-end..." + countDownLatch.getCount());
        }).start();

        String threadName = Thread.currentThread().getName();
        System.out.println(threadName + "-waiting...");

        // 等待其他线程完成
        countDownLatch.await();
        System.out.println(threadName + "-wait end...");
    }

}

7.2 多线程使用场景一：ElasticSearch 批量导入数据

在项目上线之前，我们需要把数据库中的数据一次性的同步到 ElasticSearch 索引库中，但数据量达到百万级别时，一次性读取数据的做法是不可取的（Out Of Memory）

可以使用线程池的方式导入，利用 CountDownLatch 来控制就能避免一次性加载过多，防止内存溢出

7.3 多线程使用场景二：数据汇总

在一个电商网站中，用户下单之后，需要查询数据，数据包含了三部分：订单信息、包含的商品、物流信息

这三块信息都在不同的微服务中进行实现的，我们如何完成这个业务呢

我们先来看一下常规的方案，先查询订单信息、再查询商品信息、最后查询物流信息，整个流程中每个部分是串行化执行的

我们先来看使用多线程的方案，查询订单信息、查询商品信息、查询物流信息三个操作相当于同时进行，整个流程中每个部分是并发执行的

当然，如果采用多线程的方案，需要使用 Future 接口（execute方法执行后会返回一个结果，结果的类型为 Future 接口）

7.4 多线程使用场景三：异步调用

在很多软件中都会有搜索功能，比如电商网站、地图软件等，并且这些软件会保存你的搜索记录

我们在实现搜索功能的时候，往往不会让保存搜索记录的操作影响到用户的正常搜索

我们可以选择采用异步线程来完成搜索记录的保存操作，具体要怎么操作呢？当用户开始搜索以后，我们正常返回与用户搜索内容相关的数据，用另一个线程去保存客户的搜索记录

那在代码中该如何实现呢（在 SpringBoot 项目中），只需要在保存用户搜索记录的具体方法上添加 @Async 注解

/**
 * 保存用户的搜索记录
 *
 * @param userId  Integer
 * @param keyword String
 */
@Async("taskExecutor")
@Override
public void insert(Integer userId, String keyword) {
    // 保存用户的搜索记录
    log.info("用户搜索记录保存成功,用户id:{},关键字:{}", userId, keyword);
}

同时 @Async 注解还能指定使用哪个线程池（该线程池需要由 Spring 管理）

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Configuration
public class ThreadPoolConfig {

    /**
     * 核心线程池大小
     */
    private static final int CORE_POOL_SIZE = 25;

    /**
     * 最大可创建的线程数
     */
    private static final int MAX_POOL_SIZE = 50;

    /**
     * 队列最大长度
     */
    private static final int QUEUE_CAPACITY = 1000;

    /**
     * 线程池维护线程所允许的空闲时间
     */
    private static final int KEEP_ALIVE_SECONDS = 500;

    @Bean("taskExecutor")
    public ExecutorService executorService() {
        AtomicInteger atomicInteger = new AtomicInteger(1);
        LinkedBlockingQueue<Runnable> linkedBlockingQueue = new LinkedBlockingQueue<Runnable>(QUEUE_CAPACITY);
        return new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_SECONDS,
                TimeUnit.MILLISECONDS,
                linkedBlockingQueue,
                runnable -> new Thread(runnable, "wuyanzu-pool-" + atomicInteger.getAndIncrement()),
                new ThreadPoolExecutor.DiscardPolicy()
        );
    }

}

其中 ExecutorService 类是与线程池相关的类的顶层接口（IDEA 中按下 CTRL + H 快捷键可查看继承结构）

注意事项：如果想让 @Async 注解生效，需要在 SpringBoot 的启动类上添加 @EnableAsync 注解

8. 控制某个方法允许线程并发访问的线程数量（Semaphore）

Semaphore：信号量，JUC 包下的一个工具类，实现原理基于 AQS，可以通过 Semaphore 类限制执行的线程数量

Semaphore 通常用于那些资源有明确访问数量限制的场景，常用于限流

Semaphore的使用步骤：

1. 创建 Semaphore 对象，并给定一个容量
2. semaphore.acquire()：请求一个信号量，这时候的信号量个数 - 1（一旦没有可使用的信号量，也即信号量个数变为负数时，再次请求的时候就会阻塞，直到其他线程释放了信号量）
3. semaphore.release()：释放一个信号量，此时信号量个数 + 1

代码示例：

import java.util.concurrent.Semaphore;

public class SemaphoreCase {
    public static void main(String[] args) {
        // 1.创建 semaphore 对象
        Semaphore semaphore = new Semaphore(3);

        // 2.让 10 个线程同时运行
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    // 3. 获取许可，计数 - 1
                    semaphore.acquire();
                } catch (InterruptedException interruptedException) {
                    interruptedException.printStackTrace();
                }
                try {
                    System.out.println("running...");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException interruptedException) {
                        interruptedException.printStackTrace();
                    }
                    System.out.println("end...");
                } finally {
                    // 4. 释放许可 计数 + 1
                    semaphore.release();
                }
            }).start();
        }
    }

}

9. ThreadLocal

ThreadLocal 是多线程中对于解决线程安全的一个操作类，它会为每个线程都分配一个独立的线程副本，从而解决了变量并发访问冲突的问题，ThreadLocal 同时实现了线程内的资源共享

例如，使用 JDBC 操作数据库时，会将每一个线程的 Connection 对象放入各自的 ThreadLocal 中，从而保证每个线程都在各自的 Connection 上进行数据库的操作，避免 A 线程关闭了 B 线程的连接

9.1 ThreadLocal的基本使用

ThreadLocal 的基本使用：

1. set(value)：设置值
2. get()：获取值
3. remove()：清除值

代码示例：

public class ThreadLocalTest {

    static ThreadLocal<String> threadLocal = new ThreadLocal<>();

    public static void main(String[] args) {
        new Thread(() -> {
            String threadName = Thread.currentThread().getName();
            threadLocal.set("Tom");
            removeAfterPrint(threadName);
            System.out.println(threadName + "-after remove : " + threadLocal.get());
        }, "t1").start();

        new Thread(() -> {
            String threadName = Thread.currentThread().getName();
            threadLocal.set("Jerry");
            removeAfterPrint(threadName);
            System.out.println(threadName + "-after remove : " + threadLocal.get());
        }, "t2").start();
    }

    static void removeAfterPrint(String str) {
        // 打印当前线程中本地内存中本地变量的值
        System.out.println(str + " : " + threadLocal.get());
        // 清除本地内存中的本地变量
        threadLocal.remove();
    }

}

9.2 ThreadLocal的实现原理&源码分析

ThreadLocal 本质来说就是一个线程内部存储类，让每个线程只操作自己内部的值，从而实现线程数据隔离

9.2.1 set 方法

9.2.2 get 方法

9.2.3 remove 方法

整体逻辑与 get 方法类似，找到目标元素后将其清除

9.3 ThreadLocal 的内存泄漏问题

在分析 ThreadLocal 的内存泄漏问题前，我们先来简单了解一下 Java 中的强引用和弱引用

强引用：最普通的引用方式，表示一个对象处于有用且必须的状态，如果一个对象具有强引用，则 GC 并不会回收它。即使堆内存不足，宁可抛出 OOM（Out Of Memory） 错误，也不会对其进行回收

弱引用：表示一个对象处于可能有用且非必须的状态，在 GC 线程扫描内存区域时，一旦发现弱引用，就会回收与弱引用相关联的对象。对于弱引用的回收，无论内存区域是否足够，一旦发现就会回收

每一个 Thread内部都维护一个了 ThreadLocalMap ，在 ThreadLocalMap 中的 Entry 对象继承了 WeakReference ，其中 key 为使用弱引用的 ThreadLocal 实例，value 为线程变量的副本

以下是 ThreadLocal 类的部分源码

那怎么样防止内存泄漏呢，非常简单，就是在用完 ThreadLocal 类之后，主动调用 ThreadLocal 类的 remove 方法，把数据清理掉，就能避免内存泄露的情况了