MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java 线程池的饱和策略事件

2023-01-166.7k 阅读

Java 线程池饱和策略概述

在 Java 多线程编程中，线程池是一种重要的资源管理机制，它通过复用线程来减少线程创建和销毁的开销，从而提高应用程序的性能和资源利用率。然而，当线程池的任务队列已满，并且线程池中的线程数也达到了最大限制时，新提交的任务就需要一种处理方式，这就是线程池饱和策略的作用。

Java 线程池的饱和策略定义了在线程池无法继续接受新任务时的处理逻辑。线程池的饱和策略是在 ThreadPoolExecutor 类中定义的，ThreadPoolExecutor 是 Java 线程池的核心实现类。

饱和策略接口

Java 定义了 RejectedExecutionHandler 接口来表示饱和策略，所有具体的饱和策略都需要实现这个接口。该接口只有一个方法：

public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
  • r：表示被拒绝执行的任务。
  • executor：表示当前的线程池对象。

内置饱和策略

AbortPolicy（默认策略）

当任务提交到线程池被拒绝时，AbortPolicy 会抛出 RejectedExecutionException 异常。这是 ThreadPoolExecutor 的默认饱和策略。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class AbortPolicyExample {
    public static void main(String[] args) {
        // 创建一个线程池，核心线程数为 2，最大线程数为 4，队列容量为 2
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                10,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.AbortPolicy());

        // 提交 7 个任务，超过线程池和队列的容量
        for (int i = 0; i < 7; i++) {
            executor.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

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

在上述代码中，当提交第 7 个任务时，由于线程池和队列都已满，AbortPolicy 会抛出 RejectedExecutionException 异常。

CallerRunsPolicy

CallerRunsPolicy 策略会让调用者线程（即提交任务的线程）来执行被拒绝的任务。这种策略可以减少新任务的提交速度，因为调用者线程在执行任务时，不能再提交新的任务。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class CallerRunsPolicyExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                10,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.CallerRunsPolicy());

        for (int i = 0; i < 7; i++) {
            executor.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown();
    }
}

在这个例子中，当任务超过线程池和队列容量时，主线程会执行被拒绝的任务，所以在输出中可以看到主线程也参与了任务的执行。

DiscardPolicy

DiscardPolicy 策略会直接丢弃被拒绝的任务，不会抛出任何异常，也不会执行该任务。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class DiscardPolicyExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                10,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.DiscardPolicy());

        for (int i = 0; i < 7; i++) {
            executor.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown();
    }
}

运行上述代码后，超过线程池和队列容量的任务会被直接丢弃，不会有任何输出表示这些任务被执行。

DiscardOldestPolicy

DiscardOldestPolicy 策略会丢弃任务队列中最老的一个任务（即最先进入队列的任务），然后尝试再次提交当前被拒绝的任务。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class DiscardOldestPolicyExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                10,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadPoolExecutor.DiscardOldestPolicy());

        for (int i = 0; i < 7; i++) {
            executor.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown();
    }
}

在这个示例中，当任务队列满且有新任务提交时，队列中最老的任务会被丢弃，新任务会尝试进入队列等待执行。

自定义饱和策略

除了使用 Java 内置的饱和策略，开发者还可以根据具体需求自定义饱和策略。只需实现 RejectedExecutionHandler 接口即可。

例如，我们可以创建一个记录被拒绝任务信息的自定义饱和策略：

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("任务 " + r + " 被拒绝，当前线程池状态：" + executor.toString());
    }
}

然后在创建线程池时使用这个自定义策略：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class CustomPolicyExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                10,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new CustomRejectedExecutionHandler());

        for (int i = 0; i < 7; i++) {
            executor.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown();
    }
}

在这个例子中，当任务被拒绝时，会打印出任务和线程池的相关信息，方便开发者进行调试和监控。

饱和策略的选择

在实际应用中，选择合适的饱和策略非常重要。

  • AbortPolicy：适用于对任务执行失败非常敏感的场景，比如金融交易系统。如果任务执行失败可能导致严重后果，通过抛出异常可以及时发现问题。
  • CallerRunsPolicy：适用于希望降低新任务提交速度的场景，比如在高并发情况下，通过让调用者线程执行任务，可以减缓任务提交的频率，避免系统过载。
  • DiscardPolicy：适用于对任务丢失不敏感的场景，例如日志记录任务。如果某些日志任务因为系统繁忙被丢弃，可能不会对系统核心功能造成严重影响。
  • DiscardOldestPolicy：适用于希望优先处理新任务的场景，例如实时数据处理系统，新数据可能更重要，丢弃老数据的任务可以保证新数据能及时得到处理。

自定义饱和策略则适用于那些内置策略无法满足特定业务需求的场景，开发者可以根据实际情况灵活定制处理逻辑。

线程池饱和策略与系统性能

线程池饱和策略的选择直接影响系统的性能和稳定性。不合适的饱和策略可能导致任务丢失、系统响应变慢甚至崩溃。

  • 任务丢失：如果选择 DiscardPolicy 或 DiscardOldestPolicy，在高并发情况下可能会丢失重要任务，这对于一些关键业务系统是不可接受的。
  • 系统响应变慢：CallerRunsPolicy 虽然可以减缓任务提交速度，但如果调用者线程本身是处理重要业务逻辑的线程，可能会导致业务响应变慢。
  • 系统崩溃：如果使用 AbortPolicy 且没有正确捕获 RejectedExecutionException 异常，可能会导致系统崩溃。

因此，在设计使用线程池的系统时，需要对任务的特性、系统的负载情况等进行充分的分析和测试，选择最合适的饱和策略，以确保系统的性能和稳定性。

监控与调优

为了确保线程池饱和策略能够正常工作并满足系统需求，对线程池进行监控和调优是必要的。

  • 监控线程池状态：可以通过 ThreadPoolExecutor 提供的方法获取线程池的各种状态信息，如当前线程数、队列大小、已完成任务数等。例如，getPoolSize() 方法返回当前线程池中的线程数，getQueue().size() 方法返回任务队列的大小。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolMonitoringExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                10,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2));

        executor.submit(() -> {
            System.out.println(Thread.currentThread().getName() + " 正在执行任务");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        System.out.println("当前线程池大小: " + executor.getPoolSize());
        System.out.println("任务队列大小: " + executor.getQueue().size());

        executor.shutdown();
    }
}
  • 调优线程池参数：根据监控数据，可以调整线程池的核心线程数、最大线程数和队列容量等参数。如果发现任务经常被拒绝，可以适当增加线程池的大小或队列容量；如果发现线程池中有大量空闲线程，可以适当减少核心线程数。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolTuningExample {
    public static void main(String[] args) {
        // 初始线程池参数
        int corePoolSize = 2;
        int maximumPoolSize = 4;
        long keepAliveTime = 10;
        TimeUnit unit = TimeUnit.SECONDS;
        ArrayBlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2);

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                workQueue);

        // 根据监控数据调整参数
        corePoolSize = 4;
        maximumPoolSize = 8;
        workQueue = new ArrayBlockingQueue<>(5);

        executor.setCorePoolSize(corePoolSize);
        executor.setMaximumPoolSize(maximumPoolSize);
        executor.setQueue(workQueue);

        executor.shutdown();
    }
}

通过合理的监控和调优，可以使线程池在不同的负载情况下都能保持良好的性能，并且让饱和策略发挥应有的作用。

与其他并发工具的结合使用

线程池饱和策略在与其他 Java 并发工具结合使用时，可以进一步提升系统的并发处理能力和稳定性。

  • 与 Future 结合：Future 可以用于获取任务的执行结果。当使用线程池提交任务时，可以返回一个 Future 对象。在饱和策略的处理中，如果任务被拒绝，可以根据 Future 的状态进行相应处理。
import java.util.concurrent.*;

public class FutureWithThreadPoolExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                10,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2));

        Future<Integer> future = executor.submit(() -> {
            System.out.println(Thread.currentThread().getName() + " 正在执行任务");
            Thread.sleep(2000);
            return 42;
        });

        try {
            Integer result = future.get();
            System.out.println("任务执行结果: " + result);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        executor.shutdown();
    }
}

在这个例子中，如果任务因为饱和策略被拒绝，future.get() 方法可能会抛出异常，我们可以在捕获异常时进行适当处理，比如记录日志或采取备用方案。

  • 与 CountDownLatch 结合：CountDownLatch 可以用于同步线程。假设我们有多个任务需要提交到线程池，并且需要等待所有任务完成后再进行下一步操作。在饱和策略的情况下，我们依然可以利用 CountDownLatch 来确保所有已提交的任务（包括可能被拒绝的任务）都得到妥善处理。
import java.util.concurrent.*;

public class CountDownLatchWithThreadPoolExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                10,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2));

        int taskCount = 7;
        CountDownLatch latch = new CountDownLatch(taskCount);

        for (int i = 0; i < taskCount; i++) {
            executor.submit(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + " 正在执行任务");
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    latch.countDown();
                }
            });
        }

        try {
            latch.await();
            System.out.println("所有任务已完成");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        executor.shutdown();
    }
}

在这个示例中，即使部分任务因为饱和策略被拒绝，CountDownLatch 依然可以准确地等待所有已提交任务的完成，从而保证系统逻辑的完整性。

总结线程池饱和策略的应用场景

  1. Web 服务器：在 Web 服务器处理请求时，线程池用于处理多个客户端请求。如果请求量突然剧增，超过了线程池的处理能力，合适的饱和策略可以保证服务器不会因为过载而崩溃。例如，对于一些对响应时间要求较高的 Web 应用，可以选择 CallerRunsPolicy，让调用者线程（通常是接收请求的主线程）处理部分请求，避免请求堆积导致系统瘫痪。
  2. 大数据处理：在大数据处理任务中，如 MapReduce 计算，线程池用于并行处理数据块。如果任务队列已满且线程池达到最大线程数，DiscardOldestPolicy 可能比较合适，因为新的数据块可能更具有时效性，丢弃老的数据块任务可以优先处理新数据。
  3. 消息队列系统：在消息队列系统中，线程池用于消费队列中的消息。如果消费速度跟不上生产速度，并且消息队列已经满了，AbortPolicy 可以通过抛出异常通知系统管理员，以便及时调整系统参数或增加资源。
  4. 游戏服务器：游戏服务器需要处理大量玩家的实时请求，如移动、攻击等操作。对于一些实时性要求极高的操作，如玩家的关键技能释放，使用 AbortPolicy 可以及时发现线程池饱和问题，避免因为任务丢弃导致游戏体验下降。而对于一些非关键的操作，如玩家聊天消息的处理，可以使用 DiscardPolicy，在系统繁忙时丢弃部分聊天消息，保证关键操作的正常处理。

通过深入理解和合理应用线程池饱和策略，结合与其他并发工具的协同使用以及对不同应用场景的适配，开发者可以构建出更加健壮、高效的多线程应用程序。在实际开发中，需要根据具体业务需求进行细致的分析和测试，选择最适合的饱和策略，并对线程池进行持续的监控和调优，以确保系统在各种负载情况下都能稳定运行。