多线程：2-线程池

2020-06-20 阅读量

8.线程池

线程池简介

存放线程对象的容器；事先创建指定数量的线程对象，需要时从池中取，使用完后再放入池中，不需要手动销毁线程对象。线程池减少了创建和销毁线程所带来的开销。

线程池的优点：

降低资源的消耗：通过重复利用已经创建的线程，降低了创建和销毁线程所带来的开销。
提高响应速度：当任务到达时，可以不用等待线程的创建，就能立即执行。
提高线程的可管理性：线程池可观察/调节线程的数量，优化线程等。

简单线程池设计

首先要有个线程池，其功能有：

基础功能：开启/初始化/关闭。
对外提供获取线程池中线程的功能。
要具备把线程还到线程池的功能。

加载图片！

改成更利于用户使用的线程池：
加载图片！

需要考虑的问题：

初始创建多少个线程？
没有可用线程了怎么办？
缓冲数组要多长？
缓冲数组满了怎么办？

1.线程池的核心参数

corePoolSize：核心线程数量
maximumPoolSize：最大线程数量
keepAliveTime：线程空闲后的存活时间
unit：时间单位
workQueue：用于存放任务的阻塞队列
threadFactory：线程工厂类
handler：当队列和最大线程池都满了之后的饱和策略

处理流程：
加载图片！

2.线程池可选择的阻塞队列

2.1.什么是阻塞队列？
支持阻塞插入和阻塞移除这两个附加操作的队列。

阻塞插入：当队列满时，队列会阻塞插入的线程，直到队列不满。
阻塞移除：当队列为空时，获取元素的线程会等待队列变为非空。

2.2.可选择的阻塞队列

有界队列：此队列有固定长度。
无界队列：就是此队列无限长，可一直往里追加元素。
同步移交队列：不存储元素的阻塞队列，每个插入的操作必须等到另一个线程去调用移除操作，才能成功。否则插入操作一直处于阻塞状态。

1.基于数组的有界阻塞队列：

@Test
public void arrayBlockingQueueTest() throws InterruptedException {
    // 队列最大容量为8
    ArrayBlockingQueue<Integer> queue =
        new ArrayBlockingQueue<Integer>(8);

    // 循环向队列添加值
    for(int i = 0; i < 20; i++){
        queue.put(i);
        System.out.println("向队列中添加值：" + i);
    }
}

// 结果：程序会在“向队列中添加值：7”后阻塞。

2.基于链表的有界/无界阻塞队列

@Test
public void linkedBlockingQueueTest() throws InterruptedException {
    // 队列最大容量为8（默认值为Integer.MAX_VALUE，此值非常大，可理解为是无界的阻塞队列）
    LinkedBlockingQueue<Integer> queue =
        new LinkedBlockingQueue<>(8);

    // 循环向队列添加值
    for(int i = 0; i < 20; i++){
        queue.put(i);
        System.out.println("向队列中添加值：" + i);
    }
}

// 结果：程序会在“向队列中添加值：7”后阻塞。

3.同步移阻塞队列

@Test
public void synchronousQueueTest(){
    // 没有存储元素的能力，所以这里不需要传参数
    SynchronousQueue<Integer> queue = new SynchronousQueue<>();

    // 此线程作用：向队列中插入值
    new Thread(() -> {
        try {
            queue.put(1);
            System.out.println("插入成功！");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }).start();

    // 此线程作用：从队列中移除值
    new Thread(() -> {
        try {
            queue.take();
            System.out.println("移除成功！");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }).start();
}

结果：
	移除成功！
    插入成功！

	或

	插入成功！
	移除成功！

如果把移除线程删除，那么效果就是线程卡在queue.put(1);
不继续往下执行。

3.线程池可选择的饱和策略

当阻塞队列和最大线程都已经完了的时候，再有任务提交，就需要用到饱和策略。
一共有4中饱和策略：在ThreadPoolExecutor类的源码中能找到这些策略。

AbortPolicy：终止策略（默认）
DiscardPolicy：抛弃策略（抛弃新来的任务）
DiscardOldestPolicy：抛弃旧任务策略（抛弃旧的任务）
CallerRunsPolicy：调用者运行策略

测试4种饱和策略：

环境：

// 任务类：要在线程池中执行的任务
public class Task implements Runnable {

    // 任务名称
    private String taskName;

    public Task(String taskName){
        this.taskName = taskName;
    }

    @Override
    public void run() {
        System.out.println("线程[" + Thread.currentThread().getName() + "]正在执行[" + this.taskName + "]任务");

        try {
            Thread.sleep(1000L * 5); // 模拟线程延时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("线程[" + Thread.currentThread().getName() + "]执行完[" + this.taskName + "]任务");
    }
}

3.1.AbortPolicy：终止策略（默认）

// 饱和策略
public class PolicyTest {
    // 线程池
    private static ThreadPoolExecutor executor =
            new ThreadPoolExecutor(
                    // 核心线程数，最大线程数
                    2, 3,
                    // 线程空闲后的存活时间 60s
                    60L, TimeUnit.SECONDS,
                    // 有界阻塞队列，最大容纳5个任务
                    new LinkedBlockingQueue<Runnable>(5));
    
    /**
     * 测试：测试终止策略
     *
     * 当前环境：
     * 5个任务的队列
     * 2个核心线程
     * 3个最大线程（去除掉2个核心线程，只有1个线程是空闲/可用的）
     *
     * 执行过程：
     * 1.前2个任务，会占用2个核心线程。
     * 2.第3个第7个任务，会暂存到任务队列中。
     * 3.第8个任务，会启动最大线程，去执行。
     * 4.第9、10个任务，没有线程可以去执行。启动饱和策略。
     * 5.终止策略：拒绝执行
     *
     */
    @Test
    public void abortPolicyTest() {
        // 设置线程池的饱和策略为 终止策略
        executor.setRejectedExecutionHandler(
                new ThreadPoolExecutor.AbortPolicy());

        for (int i = 1; i <= 10; i++) {
            try {
                // 提交任务到线程池中
                executor.execute(new Task("线程任务" + i));
            } catch (Exception e) {
                System.err.println(e);
            }
        }

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

    /**
     * 用于防止主线程运行完直接退出，导致我们看不到最后的结果。
     * @After：在单元测试执行完后，会执行此方法
     */
    @After
    public void after() throws InterruptedException {
        Thread.sleep(1000L * 100);
    }
}

结果：

线程[pool-1-thread-1]正在执行[线程任务1]任务        （任务1、任务2 是两个核心线程在执行）
线程[pool-1-thread-2]正在执行[线程任务2]任务
线程[pool-1-thread-3]正在执行[线程任务8]任务        （任务3、7暂存到任务队列中。任务8，在最大线程中执行）
java.util.concurrent.RejectedExecutionException:  （线程9、线程10抛出拒绝执行 的异常）
java.util.concurrent.RejectedExecutionException:
线程[pool-1-thread-1]执行完[线程任务1]任务
线程[pool-1-thread-1]正在执行[线程任务3]任务
线程[pool-1-thread-2]执行完[线程任务2]任务
线程[pool-1-thread-3]执行完[线程任务8]任务
线程[pool-1-thread-2]正在执行[线程任务4]任务
线程[pool-1-thread-3]正在执行[线程任务5]任务
线程[pool-1-thread-1]执行完[线程任务3]任务
线程[pool-1-thread-1]正在执行[线程任务6]任务
线程[pool-1-thread-2]执行完[线程任务4]任务
线程[pool-1-thread-3]执行完[线程任务5]任务
线程[pool-1-thread-2]正在执行[线程任务7]任务
线程[pool-1-thread-1]执行完[线程任务6]任务
线程[pool-1-thread-2]执行完[线程任务7]任务

3.2.DiscardPolicy：抛弃策略

在PolicyTest类中添加如下方法，其余各处保持不变。

/**
 * 测试：测试抛弃策略
 *
 * 当前环境：
 * 5个任务的队列
 * 2个核心线程
 * 3个最大线程（去除掉2个核心线程，只有1个线程是空闲/可用的）
 *
 * 执行过程：
 * 1.前2个任务，会占用2个核心线程。
 * 2.第3个第7个任务，会暂存到任务队列中。
 * 3.第8个任务，会启动最大线程，去执行。
 * 4.第9、10个任务，没有线程可以去执行。启动饱和策略。
 * 5.抛弃策略：看源码，会发现它对于接收到的任务，什么也没做（不抛异常、不执行等）。
 *                 相当于抛弃了。
 */
@Test
public void discardPolicyTest(){
    // 设置线程池的饱和策略为 抛弃策略
    executor.setRejectedExecutionHandler(
            new ThreadPoolExecutor.DiscardPolicy());

    for (int i = 1; i <= 10; i++) {
        try {
            // 提交任务到线程池中
            executor.execute(new Task("线程任务" + i));
        } catch (Exception e) {
            System.err.println(e);
        }
    }

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

结果：

线程[pool-1-thread-1]正在执行[线程任务1]任务
线程[pool-1-thread-2]正在执行[线程任务2]任务
线程[pool-1-thread-3]正在执行[线程任务8]任务
线程[pool-1-thread-2]执行完[线程任务2]任务
线程[pool-1-thread-3]执行完[线程任务8]任务
线程[pool-1-thread-1]执行完[线程任务1]任务
线程[pool-1-thread-2]正在执行[线程任务3]任务
线程[pool-1-thread-3]正在执行[线程任务5]任务
线程[pool-1-thread-1]正在执行[线程任务4]任务
线程[pool-1-thread-1]执行完[线程任务4]任务
线程[pool-1-thread-1]正在执行[线程任务6]任务
线程[pool-1-thread-2]执行完[线程任务3]任务
线程[pool-1-thread-2]正在执行[线程任务7]任务
线程[pool-1-thread-3]执行完[线程任务5]任务
线程[pool-1-thread-1]执行完[线程任务6]任务
线程[pool-1-thread-2]执行完[线程任务7]任务

3.3.DiscardOldestPolicy：抛弃旧任务策略

在PolicyTest类中添加如下方法，其余各处保持不变。

/**
 * 测试：测试抛弃旧任务策略
 *
 * 当前环境：
 * 5个任务的队列
 * 2个核心线程
 * 3个最大线程（去除掉2个核心线程，只有1个线程是空闲/可用的）
 *
 * 执行过程：
 * 1.前2个任务，会占用2个核心线程。
 * 2.第3个第7个任务，会暂存到任务队列中。
 * 3.第8个任务，会启动最大线程，去执行。
 * 4.第9、10个任务，没有线程可以去执行。启动饱和策略。
 * 5.抛弃旧任务策略：第9、10个任务分别替换掉了暂存队列中的任务3，4。
 *   相对于9、10任务来说，3、4任务是存放在暂存队列中的旧任务。
 */
@Test
public void discardOldestPolicyTest(){
    // 设置线程池的饱和策略为 抛弃旧任务策略
    executor.setRejectedExecutionHandler(
            new ThreadPoolExecutor.DiscardOldestPolicy());

    for (int i = 1; i <= 10; i++) {
        try {
            // 提交任务到线程池中
            executor.execute(new Task("线程任务" + i));
        } catch (Exception e) {
            System.err.println(e);
        }
    }

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

结果：会发现任务3、4被抛弃了。

线程[pool-1-thread-1]正在执行[线程任务1]任务
线程[pool-1-thread-2]正在执行[线程任务2]任务
线程[pool-1-thread-3]正在执行[线程任务8]任务
线程[pool-1-thread-1]执行完[线程任务1]任务
线程[pool-1-thread-2]执行完[线程任务2]任务
线程[pool-1-thread-3]执行完[线程任务8]任务
线程[pool-1-thread-1]正在执行[线程任务5]任务
线程[pool-1-thread-3]正在执行[线程任务7]任务
线程[pool-1-thread-2]正在执行[线程任务6]任务
线程[pool-1-thread-1]执行完[线程任务5]任务
线程[pool-1-thread-3]执行完[线程任务7]任务
线程[pool-1-thread-2]执行完[线程任务6]任务
线程[pool-1-thread-3]正在执行[线程任务10]任务
线程[pool-1-thread-1]正在执行[线程任务9]任务
线程[pool-1-thread-3]执行完[线程任务10]任务
线程[pool-1-thread-1]执行完[线程任务9]任务

3.4.CallerRunsPolicy：调用者运行策略

在PolicyTest类中添加如下方法，其余各处保持不变。

/**
 * 测试：测试调用者运行策略
 *
 * 当前环境：
 * 5个任务的队列
 * 2个核心线程
 * 3个最大线程（去除掉2个核心线程，只有1个线程是空闲/可用的）
 *
 * 执行过程：
 * 1.前2个任务，会占用2个核心线程。
 * 2.第3个第7个任务，会暂存到任务队列中。
 * 3.第8个任务，会启动最大线程，去执行。
 * 4.第9、10个任务，没有线程可以去执行。启动饱和策略。
 * 5.调用者运行策略：没有可用的线程，由主线程（调用者）来执行第9、10个任务。
 * （若线程池没有关闭，当前的任务对象直接执行run方法，相当于调用方自己在执行任务）
 */
@Test
public void callerRunsPolicyTest(){
    // 设置线程池的饱和策略为 调用者运行策略
    executor.setRejectedExecutionHandler(
            new ThreadPoolExecutor.CallerRunsPolicy());

    for (int i = 1; i <= 10; i++) {
        try {
            // 提交任务到线程池中
            executor.execute(new Task("线程任务" + i));
        } catch (Exception e) {
            System.err.println(e);
        }
    }

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

结果：

线程[main]正在执行[线程任务9]任务
线程[pool-1-thread-1]正在执行[线程任务1]任务
线程[pool-1-thread-3]正在执行[线程任务8]任务
线程[pool-1-thread-2]正在执行[线程任务2]任务
线程[main]执行完[线程任务9]任务
线程[main]正在执行[线程任务10]任务
线程[pool-1-thread-3]执行完[线程任务8]任务
线程[pool-1-thread-1]执行完[线程任务1]任务
线程[pool-1-thread-1]正在执行[线程任务4]任务
线程[pool-1-thread-2]执行完[线程任务2]任务
线程[pool-1-thread-3]正在执行[线程任务3]任务
线程[pool-1-thread-2]正在执行[线程任务5]任务
线程[main]执行完[线程任务10]任务
线程[pool-1-thread-1]执行完[线程任务4]任务
线程[pool-1-thread-2]执行完[线程任务5]任务
线程[pool-1-thread-2]正在执行[线程任务7]任务
线程[pool-1-thread-3]执行完[线程任务3]任务
线程[pool-1-thread-1]正在执行[线程任务6]任务
线程[pool-1-thread-1]执行完[线程任务6]任务
线程[pool-1-thread-2]执行完[线程任务7]任务

4.线程池示意图

1.线程池的执行示意图：
加载图片！

主线程调用execute() 方法，执行一个线程任务：

如果核心线程池没满，会走①，立即创建一个线程，来执行这个任务。
如果核心线程池满了，会走②，将这个任务提交到阻塞队列里。核心线程会池里的线程会不断轮询阻塞队列，拿到新的任务来执行。
如果核心线程池和阻塞队列都满了，会走③，会创建新的线程（前提是线程池中的线程数，未超过最大线程数）
如果核心线程池、阻塞队列、最大线程池都满了，会走④，调用饱和策略。根据构造线程池时传入的不同的饱和策略，执行相应的处理。如果使用的饱和策略是CallerRunsPolicy，会接着走⑤，返回到主线程，通过主线程去运行任务的run方法来执行任务。

5.常用线程池（创建线程池的3种方式）

5.1.newCachedThreadPool

/**
 * 线程池数量无限的线程池
 */
public static ExecutorService newCachedThreadPool() {
    // 核心线程数0，最大线程数Integer.MAX_VALUE（相当于无限大）
    // keepAliveTime为60s
    // 默认阻塞队列是同步移交阻塞队列。
    // 有一个任务，就对应着有一个线程去消费，才能接收另一个任务，又因为最大
    // 线程数为无限大，所以此线程池是 线程池数量无限的线程池。
    return ThreadPoolExecutor(0, Integer.MAX_VALUE,
                             60L, TimeUnit.SECONDS,
                             new SynchronousQueue<Runnable>());
}

// 注意：因为线程数量无限大，那么此线程池就有创建无限个线程的可能，从而导致系统崩溃。

5.2.newFixedThreadPool

/**
 * 线程池数量固定的线程池
 */
public static ExecutorService newFixedThreadPool(int nThreads) {
    // 核心线程数和最大线程数均为nThreads，既是初始值又是最大值，所以线程数量固定。
    // keepAliveTime为0ms
    // 默认阻塞队列是无参数的无界阻塞队列。
    // 线程数量固定，阻塞队列无限大，所以此线程池是 线程池数量固定的线程池。
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

// 注意：线程个数虽然固定，但是阻塞队列是无限制的，如果提交的任务过多，队列也会把内存撑爆。

5.3.newSingleThreadExecutor

/**
 * 单一线程的线程池
 */
public static ExecutorService newSingleThreadExecutor() {
    // 核心线程数和最大线程数均为 1 。
    // keepAliveTime为0ms
    // 默认阻塞队列是无参数的无界阻塞队列。
    // 线程数量为1，阻塞队列无限大，所以此线程池是 单一线程的线程池。
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

5.4.newScheduledThreadPool

1
2
3

/**
 * 适用于执行延迟或周期性任务
 */

6.向线程池提交任务两种方式

方式一：利用submit方法提交任务，并接收任务的返回结果。

@Test
public void submitTest() throws Exception {
    // 创建线程池
    ExecutorService threadPool =
            Executors.newCachedThreadPool();

    //向线程池提交任务方式一：利用submit方法提交任务，并接收任务的返回结果。
    Future<Integer> future = threadPool.submit(() -> {
        Thread.sleep(1000L * 10);  // 模拟线程延时
        return 2 * 5;
    });

    // get() 是一个阻塞方法，直到任务有返回值之后，才会向下执行。
    Integer num = future.get();

    System.out.println("执行结果：" + num);
}

// 结果：延迟10s后，打印“执行结果：10”

方式二：利用execute方法提交任务，没有返回结果。

@Test
public void executeTest() throws Exception {
    // 创建线程池
    ExecutorService threadPool =
            Executors.newCachedThreadPool();

    //向线程池提交任务方式二：利用execute方法提交任务，没有返回结果。
    threadPool.execute(() -> {

        try {
            Thread.sleep(1000L * 10);  // 模拟线程延时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        Integer num = 2 * 5;
        System.out.println("执行结果：" + num);
    });

    Thread.sleep(1000L * 20);  // 主线程延时，以便拿到打印的结果
}

// 结果：延迟10s后，打印“执行结果：10”

7.线程池的状态：

加载图片！