多线程底层——自己动手写一个线程池

见解分享

2024-01-17 01:07:05

打造自己的线程池：深入剖析其底层实现

线程池：并发编程的利器

在现代软件开发中，并发编程已成为提升程序效率和响应能力的必备手段。线程池作为一种管理线程并发的高效机制，在各种场景下发挥着至关重要的作用。本文将带领你深入线程池的底层实现，亲自打造一个功能完善的线程池，让你全面掌握其工作原理和核心组件。

线程池的基础

线程池本质上是一个管理线程的容器，它维护着一组可复用的线程，用于处理任务。与直接创建线程相比，线程池具有以下优势：

资源复用： 线程池中的线程可以重复使用，避免频繁创建和销毁线程带来的开销。
性能优化： 预先创建和管理线程可以减少任务处理的延迟，大幅提升程序效率。
并发控制： 线程池可以限制并发线程的数量，防止系统因线程过多而崩溃。

ThreadPoolExecutor 的实现

Java 并发包中的 ThreadPoolExecutor 类提供了线程池的标准实现。它定义了一系列参数和方法，用于配置和管理线程池的行为。接下来，我们将基于 ThreadPoolExecutor 的思路，构建我们自己的线程池：

核心组件

一个线程池主要由以下组件组成：

任务队列： 用于存储待执行的任务。
工作线程： 用于执行任务的线程。
拒绝策略： 当任务队列已满，无法再接收新任务时，用于处理拒绝任务的策略。

自定义线程池

1. 任务队列

任务队列是线程池的关键组成部分，它决定了任务的排队和执行顺序。我们可以根据实际需求选择不同的队列实现，例如 ArrayBlockingQueue 或 LinkedBlockingQueue。

private BlockingQueue<Runnable> taskQueue;

2. 工作线程

工作线程是线程池的核心，负责执行任务。每个工作线程都以守护线程的方式运行，一旦所有任务执行完毕，线程池就会自动关闭。

private ExecutorService executor;

3. 拒绝策略

当任务队列已满时，拒绝策略决定了如何处理新任务。常用的拒绝策略包括：

AbortPolicy：抛出 RejectedExecutionException 异常。
DiscardPolicy：直接丢弃任务。
DiscardOldestPolicy：丢弃队列中最早的任务。

private RejectedExecutionHandler rejectedExecutionHandler;

4. 线程池初始化

public CustomThreadPoolExecutor(int corePoolSize,
                               int maximumPoolSize,
                               long keepAliveTime,
                               TimeUnit unit,
                               BlockingQueue<Runnable> workQueue,
                               ThreadFactory threadFactory,
                               RejectedExecutionHandler handler) {
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.keepAliveTime = keepAliveTime;
    this.unit = unit;
    this.workQueue = workQueue;
    this.threadFactory = threadFactory;
    this.rejectedExecutionHandler = handler;
    prestartAllCoreThreads();
}

核心方法

除了核心组件，线程池还提供了以下重要方法：

execute(Runnable task)： 向线程池提交一个任务。
shutdown()： 关闭线程池，阻止接收新任务。
awaitTermination()： 等待线程池关闭完成。

自定义线程池使用示例

CustomThreadPoolExecutor executor = new CustomThreadPoolExecutor(
    4, // 核心线程数
    8, // 最大线程数
    0, // 空闲线程存活时间
    TimeUnit.MILLISECONDS, // 时间单位
    new ArrayBlockingQueue<>(10), // 任务队列
    new DefaultThreadFactory(), // 线程工厂
    new ThreadPoolExecutor.CallerRunsPolicy()); // 拒绝策略

executor.execute(() -> {
    // 任务代码
});

executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);