深入解析Java线程池的扩容机制与拒绝策略-阿里云开发者社区

深入解析Java线程池的扩容机制与拒绝策略

Java线程池是多线程编程中一项重要的工具，它能够有效地管理和调度线程，提高程序的并发性能。线程池的扩容机制是线程池的关键特性之一，它允许根据工作负载的变化动态地增加或减少线程数量。

背景

在并发编程中，创建和销毁线程是一项开销较大的操作。为了更有效地利用系统资源，避免不必要的线程创建和销毁，Java引入了线程池的概念。线程池通过预先创建一定数量的线程，并将它们保存在池中，可以在需要时重用这些线程，避免频繁地创建和销毁线程。为了适应动态的工作负载，线程池的扩容机制应运而生。

线程池扩容原理

Java线程池的扩容机制基于任务队列中任务的数量。当任务队列中的任务数量达到一定阈值时，线程池会动态地增加线程数量。这样可以确保足够的线程用于处理任务，提高系统的并发性能。

主要参数

在深入讨论扩容机制之前，我们先了解一下线程池的主要参数：

corePoolSize（核心线程数）：线程池中一直存活的线程数量，即使它们处于空闲状态也不会被销毁。
maximumPoolSize（最大线程数）：线程池中允许的最大线程数量，包括空闲状态的线程和正在执行任务的线程。
workQueue（工作队列）：存放等待执行的任务的队列。当任务提交到线程池时，会先放入工作队列。
keepAliveTime（线程空闲时间）：空闲线程的最大存活时间，超过这个时间空闲线程将被销毁，仅当线程数量超过核心线程数时生效。
RejectedExecutionHandler（任务拒绝处理器）：当任务无法被执行时的处理策略。

扩容触发条件

线程池的扩容通常有两个触发条件：

任务队列满：当任务队列中的任务数量超过一定阈值时，即 workQueue 的容量达到上限。
任务无法及时处理：当线程池中的线程数达到 maximumPoolSize 时，新提交的任务无法被及时处理。

扩容流程

当扩容触发条件满足时，线程池会按照以下流程进行扩容：

判断是否达到最大线程数：如果当前线程数量未达到 maximumPoolSize，则继续扩容。
创建新线程：创建新的线程，并将其添加到线程池中。
执行任务：新创建的线程会立即执行工作队列中的任务。
线程数量动态调整：在任务执行完成后，线程池可能会动态调整线程数量。如果空闲线程超过了 keepAliveTime，则可能会被销毁，以节省资源。

对于java来说，如果认为数量大于核心线程数之后，那么要么扩容核心线程数，要么让非核心线程处理任务。

在Java线程池中，当任务数量大于核心线程数时，线程池有两种基本策略来处理任务：

扩容核心线程数：线程池会尝试创建新的核心线程来处理任务。这有助于更快地响应任务，尤其是在任务到达时，避免等待非核心线程的启动。
让非核心线程处理任务：如果核心线程数已满，线程池会将多余的任务放入工作队列中，由非核心线程来处理。非核心线程在工作队列中取出任务执行，这样可以在一定程度上避免频繁地创建和销毁线程，节省资源。

具体实现方式

Java线程池的扩容机制在 ThreadPoolExecutor 类中得到了具体的实现。以下是一个简单的案例，演示了如何创建一个具备扩容机制的线程池：

import java.util.concurrent.*;
public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,  // corePoolSize
                5,  // maximumPoolSize
                1,  // keepAliveTime
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(10)  // workQueue
        );
        // 提交任务
        for (int i = 1; i <= 20; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " executed by " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000);  // 模拟任务执行时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        // 关闭线程池
        executor.shutdown();
    }
}

在这个例子中，线程池的核心线程数为2，最大线程数为5，工作队列容量为10。当任务数量超过线程池的当前线程数量时，线程池会自动扩容，创建新的线程来处理任务。

场景描述

考虑一个简单的网络爬虫应用，该应用从网页上下载图片并进行处理。爬虫需要处理的任务量不确定，取决于要爬取的网页数量。在这种情况下，使用线程池可以有效地管理和调度任务，提高并发处理能力。

解决方案

我们将创建一个线程池，并利用其扩容机制，以适应动态的任务负载。在任务提交过程中，观察线程池的行为，特别是当任务数量超过核心线程数和最大线程数时，线程池是如何扩容的。

import java.util.concurrent.*;
public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,  // corePoolSize
                5,  // maximumPoolSize
                1,  // keepAliveTime
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(10)  // workQueue
        );
        // 提交任务
        for (int i = 1; i <= 20; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " executed by " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000);  // 模拟任务执行时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        // 关闭线程池
        executor.shutdown();
    }
}

分析与观察

核心线程数：2。线程池初始化时创建了2个核心线程，它们会一直存活，即使是空闲状态。
最大线程数：5。最大线程数表示线程池允许的最大线程数量，包括核心线程和非核心线程。在本例中，最大线程数为5，超过这个数量的任务将放入工作队列中。
工作队列容量：10。工作队列是一个有限容量的阻塞队列，用于存放等待执行的任务。在本例中，工作队列容量为10。

观察输出

执行上述代码后，我们观察到线程池的行为：

当前有2个核心线程被创建，分别执行任务1和任务2。
当任务提交数量增加（任务3至任务11），这些任务被放入工作队列中，等待核心线程执行。
当任务数量超过核心线程数，新的任务触发了线程池的扩容机制。此时创建了非核心线程，开始执行任务。
当非核心线程执行完任务后，空闲一段时间（根据 keepAliveTime 的设置），非核心线程可能被销毁，以减少线程池中的线程数量。
执行剩余的任务，观察线程的创建和销毁。

结果分析

通过观察输出，我们可以发现线程池在任务负载增加时，动态地创建新线程以处理任务，并在任务处理完成后，通过一定的策略（例如空闲线程销毁策略）来调整线程数量，从而更好地适应不同的工作负载。

任务拒绝处理和拒绝策略

在使用线程池时，当任务的数量超过线程池的处理能力，就会触发任务拒绝处理。任务拒绝处理是指当无法再接受新任务时，线程池会采取一种策略来处理这些被拒绝的任务。Java线程池提供了灵活的拒绝策略机制，允许开发者自定义拒绝策略。

常见的拒绝策略

AbortPolicy（默认策略）：该策略会直接抛出 RejectedExecutionException 异常，通知调用者线程池已满，无法接受新任务。
CallerRunsPolicy：当任务被拒绝时，由提交任务的线程（调用线程）执行该任务。这样一来，虽然不能异步执行任务，但至少不会丢失任务。
DiscardPolicy：当任务被拒绝时，直接丢弃掉这个任务，没有任何处理。
DiscardOldestPolicy：当任务被拒绝时，丢弃工作队列中最旧的任务，然后尝试重新提交被拒绝的任务。

拒绝策略比喻解释

考虑一个场景：你是一家热门餐厅的厨师，你负责配发任务的，正常情况下自己不做，让别人做，而你的厨房容量有限，同时有很多顾客提交了菜品的订单。这里的顾客订单就相当于线程池中的任务。

AbortPolicy（默认策略）：

比喻：你的厨房里只有有限的炉灶，当订单太多，超过炉灶的处理能力时，你选择不接受新的订单。
解释：这就像线程池告诉提交任务的线程说，“抱歉，我们无法接受更多的任务了，因为线程池已经满了。”并抛出一个异常，类似于“我们已经无法处理更多的订单了”。

CallerRunsPolicy：

比喻：当订单太多，你的炉灶无法处理时，你决定亲自下厨，处理一部分订单，虽然这样做会增加你的工作负担。
解释：这就像线程池告诉提交任务的线程说，“虽然线程池满了，但是你可以自己来执行这个任务，不过要注意，这样做可能会增加你的负担。”

DiscardPolicy：

比喻：当订单太多，而你已经无法应付时，你选择直接忽略掉一部分订单，不做处理。
解释：这就像线程池直接丢弃掉无法处理的任务，不做任何处理，就好像这些任务从来没有提交过一样。

DiscardOldestPolicy：

比喻：当订单太多，而你无法处理新订单时，你选择丢弃掉等待时间最长的订单，然后接受新订单。
解释：这就像线程池告诉提交任务的线程说，“我们丢弃掉队列中等待时间最长的任务，然后尝试重新提交新的任务。”这样做可能会让一些任务被“踢出”队列，但至少能够接受新的任务。

自定义拒绝策略

除了使用预定义的拒绝策略，我们还可以自定义拒绝策略。为此，需要实现 RejectedExecutionHandler 接口，并实现其 rejectedExecution 方法。以下是一个简单的自定义拒绝策略的例子：

import java.util.concurrent.*;
public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 自定义处理逻辑
        System.err.println("Task rejected. " + r.toString());
    }
}

然后，在创建线程池时，将自定义的拒绝策略传递给 ThreadPoolExecutor 的构造方法：

ThreadPoolExecutor executor = new ThreadPoolExecutor(
        2,                    // corePoolSize
        5,                    // maximumPoolSize
        1,                    // keepAliveTime
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(10),  // workQueue
        new CustomRejectedExecutionHandler()  // 自定义拒绝策略
);

案例分析

考虑以下情景：一个在线电商平台，用户可以提交商品评论。评论的处理需要通过线程池来异步进行，以提高用户体验。我们希望在评论处理的线程池中使用自定义的拒绝策略，以便更好地处理高并发的情况。

import java.util.concurrent.*;
public class CommentProcessor {
    public static void main(String[] args) {
        // 创建线程池，使用自定义的拒绝策略
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,  // corePoolSize
                5,  // maximumPoolSize
                1,  // keepAliveTime
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(10),  // workQueue
                new CustomRejectedExecutionHandler()  // 自定义拒绝策略
        );
        // 模拟用户提交评论
        for (int i = 1; i <= 20; i++) {
            final int commentId = i;
            executor.submit(() -> {
                System.out.println("Processing comment " + commentId + " by " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000);  // 模拟评论处理时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        // 关闭线程池
        executor.shutdown();
    }
}

结果分析

执行上述代码后，我们观察到：

当线程池的工作队列满了，且无法继续创建新线程时，会触发自定义的拒绝策略。
自定义拒绝策略在 rejectedExecution 方法中输出了被拒绝的任务信息，这使得开发者能够更好地了解拒绝的原因。

如果大家觉得有用的话，可以关注我下面的微信公众号，极客李华，我会在里面更新更多行业资讯，企业面试内容，编程资源，如何写出可以让大厂面试官眼前一亮的简历，让大家更好学习编程，我的抖音，B站也叫极客李华。

深入解析Java线程池的扩容机制与拒绝策略