发布于 由admin

JAVA ThreadPoolExecutor 拒绝策略触发过早？线程池尺寸与队列调优技巧

JAVA ThreadPoolExecutor 拒绝策略触发过早？线程池尺寸与队列调优技巧

各位听众，大家好！今天我们来深入探讨一个在使用 ThreadPoolExecutor 时经常遇到的问题：拒绝策略触发过早。很多开发者在使用线程池时，会发现即使线程池还有空闲线程或者队列尚未满，拒绝策略却已经开始生效，导致任务被拒绝。这背后到底是什么原因？我们又该如何正确地配置线程池的尺寸和队列，才能避免此类问题的发生，并最大限度地提升应用的性能呢？

1. 理解 ThreadPoolExecutor 的工作原理

要理解拒绝策略触发过早的原因，首先必须彻底理解 ThreadPoolExecutor 的工作原理。ThreadPoolExecutor 的执行流程大致如下：

  1. 任务提交: 当我们通过 executor.execute(Runnable task) 提交任务时，线程池会尝试执行该任务。
  2. 线程创建:
    • 如果当前线程池中的线程数小于 corePoolSize，则线程池会创建一个新的线程来执行该任务。
    • 如果当前线程池中的线程数已经达到 corePoolSize，则任务会被放入阻塞队列 workQueue 中等待执行。
  3. 队列排队: 任务在 workQueue 中等待，直到有空闲线程可用。
  4. 线程复用: 当有空闲线程时，它会从 workQueue 中取出一个任务来执行。
  5. 线程扩容:
    • 如果 workQueue 已满，并且当前线程池中的线程数小于 maximumPoolSize，则线程池会创建一个新的线程来执行该任务。
    • 如果 workQueue 已满，并且当前线程池中的线程数已经达到 maximumPoolSize，则会触发拒绝策略。

这个流程中，几个关键参数决定了线程池的行为：

  • corePoolSize: 核心线程数，即线程池中始终保持的线程数量。即使没有任务，这些线程也不会被销毁。
  • maximumPoolSize: 最大线程数，即线程池中允许存在的最大线程数量。
  • keepAliveTime: 线程空闲存活时间，当线程池中的线程数超过 corePoolSize 时，多余的空闲线程会在指定时间内被销毁。
  • workQueue: 阻塞队列，用于存放等待执行的任务。常见的阻塞队列有 ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue 和 SynchronousQueue。
  • RejectedExecutionHandler: 拒绝策略，当线程池无法处理新提交的任务时，会调用该策略来处理被拒绝的任务。常见的拒绝策略有 AbortPolicy、CallerRunsPolicy、DiscardPolicy 和 DiscardOldestPolicy。

2. 拒绝策略触发过早的常见原因

理解了 ThreadPoolExecutor 的工作原理后，我们就可以分析拒绝策略触发过早的常见原因：

  • corePoolSize 过小: 如果 corePoolSize 设置得太小，即使系统资源充足，线程池也只会创建少量线程来执行任务，导致大量任务堆积在 workQueue 中。当 workQueue 满后，新的任务就会被拒绝。
  • workQueue 容量过小: 如果 workQueue 的容量设置得太小，即使 corePoolSize 足够大，workQueue 也会很快被填满，导致新的任务被拒绝。
  • 选择了不合适的 workQueue 实现: 不同的 workQueue 实现有不同的特性。例如，SynchronousQueue 不存储任务，每个插入操作必须等待一个相应的移除操作，因此通常需要与足够大的 maximumPoolSize 配合使用。如果 maximumPoolSize 和 corePoolSize 相等，并且使用了 SynchronousQueue，那么任何超出 corePoolSize 的任务都会立即被拒绝。
  • 任务执行时间过长: 如果任务的执行时间过长，线程池中的线程会被长时间占用，导致新的任务无法得到及时处理，从而导致 workQueue 迅速填满。
  • 任务提交速率过快: 如果任务的提交速率远大于线程池的处理能力，即使线程池的配置看起来合理，workQueue 仍然可能被迅速填满，导致拒绝策略生效。

3. 代码示例：复现拒绝策略触发过早的问题

为了更直观地理解这些原因，我们来看几个代码示例。

示例 1：corePoolSize 过小

import java.util.concurrent.*;

public class RejectedExecutionExample1 {

    public static void main(String[] args) {
        int corePoolSize = 2;
        int maximumPoolSize = 4;
        long keepAliveTime = 60L;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(10); // 队列容量为10
        RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                TimeUnit.SECONDS,
                workQueue,
                rejectedExecutionHandler
        );

        for (int i = 0; i < 20; i++) {
            final int taskNumber = i;
            try {
                executor.execute(() -> {
                    System.out.println("Executing task " + taskNumber + " by thread " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(1000); // 模拟任务执行时间
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });
            } catch (RejectedExecutionException e) {
                System.err.println("Task " + taskNumber + " rejected: " + e.getMessage());
            }
        }

        executor.shutdown();
    }
}

在这个例子中，corePoolSize 设置为 2，workQueue 的容量为 10。这意味着只有 2 个线程会立即执行任务，其余的任务会被放入队列。当提交的任务数量超过 12 个时（2 个线程正在执行 + 10 个任务在队列中），就会触发 AbortPolicy，导致 RejectedExecutionException。

示例 2：workQueue 容量过小

import java.util.concurrent.*;

public class RejectedExecutionExample2 {

    public static void main(String[] args) {
        int corePoolSize = 4;
        int maximumPoolSize = 8;
        long keepAliveTime = 60L;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2); // 队列容量为2
        RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                TimeUnit.SECONDS,
                workQueue,
                rejectedExecutionHandler
        );

        for (int i = 0; i < 20; i++) {
            final int taskNumber = i;
            try {
                executor.execute(() -> {
                    System.out.println("Executing task " + taskNumber + " by thread " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(1000); // 模拟任务执行时间
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });
            } catch (RejectedExecutionException e) {
                System.err.println("Task " + taskNumber + " rejected: " + e.getMessage());
            }
        }

        executor.shutdown();
    }
}

在这个例子中，corePoolSize 设置为 4，但 workQueue 的容量只有 2。这意味着只有 4 个线程会立即执行任务，其余的任务会被放入队列。当提交的任务数量超过 6 个时（4 个线程正在执行 + 2 个任务在队列中），就会触发 AbortPolicy。

示例 3：使用 SynchronousQueue

import java.util.concurrent.*;

public class RejectedExecutionExample3 {

    public static void main(String[] args) {
        int corePoolSize = 4;
        int maximumPoolSize = 4; // corePoolSize 和 maximumPoolSize 相等
        long keepAliveTime = 60L;
        BlockingQueue<Runnable> workQueue = new SynchronousQueue<>(); // 使用 SynchronousQueue
        RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                TimeUnit.SECONDS,
                workQueue,
                rejectedExecutionHandler
        );

        for (int i = 0; i < 20; i++) {
            final int taskNumber = i;
            try {
                executor.execute(() -> {
                    System.out.println("Executing task " + taskNumber + " by thread " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(1000); // 模拟任务执行时间
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });
            } catch (RejectedExecutionException e) {
                System.err.println("Task " + taskNumber + " rejected: " + e.getMessage());
            }
        }

        executor.shutdown();
    }
}

在这个例子中，corePoolSize 和 maximumPoolSize 都设置为 4，并且使用了 SynchronousQueue。由于 SynchronousQueue 不存储任务，因此任何超出 corePoolSize 的任务都会立即被拒绝。

4. 线程池尺寸与队列的调优技巧

那么，如何正确地配置线程池的尺寸和队列，才能避免拒绝策略触发过早的问题，并最大限度地提升应用的性能呢？

4.1 确定 corePoolSize 和 maximumPoolSize

corePoolSize 和 maximumPoolSize 的选择取决于任务的类型和系统的资源状况。

  • CPU 密集型任务: 对于 CPU 密集型任务，线程数通常设置为 CPU 核心数 + 1。多出的一个线程可以减少线程上下文切换带来的开销。可以使用 Runtime.getRuntime().availableProcessors() 获取 CPU 核心数。
  • IO 密集型任务: 对于 IO 密集型任务，线程数可以设置得比 CPU 核心数大得多，因为线程在等待 IO 操作时不会占用 CPU 资源。通常，可以将线程数设置为 CPU 核心数的 2 倍甚至更多。
  • 混合型任务: 对于混合型任务，需要根据 CPU 密集型和 IO 密集型任务的比例进行调整。

maximumPoolSize 的设置应该考虑系统的资源限制，避免线程数过多导致系统崩溃。通常，maximumPoolSize 应该大于或等于 corePoolSize。

4.2 选择合适的 workQueue 实现

不同的 workQueue 实现有不同的特性，适用于不同的场景。

workQueue 类型 特点 适用场景
ArrayBlockingQueue 基于数组的有界阻塞队列，按照 FIFO（先进先出）原则处理任务。 适用于任务数量有限，需要控制任务执行顺序的场景。
LinkedBlockingQueue 基于链表的无界阻塞队列（也可以指定容量），按照 FIFO 原则处理任务。 适用于任务数量不确定，但需要保证任务不会被拒绝的场景。注意，如果任务提交速率过快，无界队列可能会导致内存溢出。
PriorityBlockingQueue 具有优先级的无界阻塞队列，任务按照优先级顺序执行。 适用于需要根据任务的优先级来调度任务的场景。
SynchronousQueue 不存储任务的阻塞队列，每个插入操作必须等待一个相应的移除操作。 适用于任务提交速率和处理速率相近的场景，或者需要快速将任务传递给可用线程的场景。通常需要与足够大的 maximumPoolSize 配合使用，否则容易触发拒绝策略。
DelayQueue 延迟队列，任务只有在指定的延迟时间到达后才能被执行。 适用于需要延迟执行任务的场景，例如定时任务、缓存过期等。

4.3 监控线程池状态

为了更好地了解线程池的运行状况，我们需要对其进行监控。ThreadPoolExecutor 提供了一些方法来获取线程池的状态信息：

  • getPoolSize(): 获取线程池中的线程数。
  • getActiveCount(): 获取正在执行任务的线程数。
  • getQueue().size(): 获取等待执行的任务数。
  • getCompletedTaskCount(): 获取已完成的任务数。
  • getTaskCount(): 获取已提交的任务总数。

通过监控这些指标，我们可以及时发现线程池的瓶颈，并进行相应的调整。

4.4 选择合适的 RejectedExecutionHandler

RejectedExecutionHandler 用于处理被拒绝的任务。常见的拒绝策略有：

  • AbortPolicy: 默认策略，直接抛出 RejectedExecutionException。
  • CallerRunsPolicy: 由提交任务的线程来执行被拒绝的任务。
  • DiscardPolicy: 直接丢弃被拒绝的任务，不进行任何处理。
  • DiscardOldestPolicy: 丢弃队列中最老的任务，然后尝试重新提交被拒绝的任务。

选择哪种拒绝策略取决于应用的具体需求。如果任务非常重要，不能丢失，可以选择 CallerRunsPolicy 或者自定义拒绝策略，将任务重新放入队列或者保存到数据库中。如果任务不重要，可以选择 DiscardPolicy。

4.5 动态调整线程池参数

在某些情况下，我们需要根据系统的负载情况动态调整线程池的参数。例如，当系统负载较高时，可以增加 corePoolSize 和 maximumPoolSize，以提高线程池的处理能力。当系统负载较低时，可以减少 corePoolSize，以节省系统资源。

动态调整线程池参数可以使用 setCorePoolSize() 和 setMaximumPoolSize() 方法。

5. 代码示例：线程池调优

我们来看一个完整的线程池调优的例子。

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadPoolTuningExample {

    private static final int CPU_CORES = Runtime.getRuntime().availableProcessors();
    private static final int CORE_POOL_SIZE = CPU_CORES;
    private static final int MAXIMUM_POOL_SIZE = CPU_CORES * 2;
    private static final long KEEP_ALIVE_TIME = 60L;
    private static final int QUEUE_CAPACITY = 100;

    public static void main(String[] args) throws InterruptedException {
        // 使用自定义的线程工厂，方便命名线程
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setNameFormat("task-pool-%d")
                .build();

        // 使用 ArrayBlockingQueue，限制队列大小
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);

        // 使用 CallerRunsPolicy，避免任务丢失
        RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();

        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAXIMUM_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                workQueue,
                threadFactory,
                rejectedExecutionHandler
        );

        // 预热核心线程
        executor.prestartAllCoreThreads();

        // 模拟任务提交
        for (int i = 0; i < 200; i++) {
            final int taskNumber = i;
            executor.execute(() -> {
                System.out.println("Executing task " + taskNumber + " by thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(500); // 模拟任务执行时间
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });

            // 每隔一段时间输出线程池状态
            if (i % 20 == 0) {
                System.out.println("Pool Size: " + executor.getPoolSize() +
                        ", Active Threads: " + executor.getActiveCount() +
                        ", Queue Size: " + executor.getQueue().size() +
                        ", Completed Tasks: " + executor.getCompletedTaskCount());
            }
        }

        // 等待所有任务完成
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);

        System.out.println("All tasks finished.");
    }

    // 自定义线程工厂，方便命名线程
    static class ThreadFactoryBuilder implements ThreadFactory {
        private String nameFormat = null;
        private final AtomicInteger poolNumber = new AtomicInteger(1);
        private final AtomicInteger threadNumber = new AtomicInteger(1);

        ThreadFactoryBuilder setNameFormat(String nameFormat) {
            this.nameFormat = nameFormat;
            return this;
        }

        @Override
        public Thread newThread(Runnable runnable) {
            String threadName = String.format(nameFormat, threadNumber.getAndIncrement());
            Thread thread = new Thread(runnable, threadName);
            return thread;
        }
    }
}

在这个例子中，我们根据 CPU 核心数设置了 corePoolSize 和 maximumPoolSize，使用了 ArrayBlockingQueue 来限制队列大小，并使用了 CallerRunsPolicy 来避免任务丢失。同时，我们还使用了自定义的线程工厂来方便命名线程，并定期输出线程池的状态信息。

6.总结一些关键点

  • 线程池配置不当，如过小的 corePoolSize 或 workQueue，容易导致拒绝策略过早触发。
  • 监控线程池状态，根据任务类型和系统资源合理选择线程池参数，是提升应用性能的关键。
  • 不同 workQueue 和 RejectedExecutionHandler 适用于不同场景，要根据实际需求选择合适的实现。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注