最近的项目里用到了线程池来处理“生产者-消费者”模式，今天主要是来系统学习一下线程池的概念以及如何自己手搓一个简易的线程池。加深一下对线程池使用策略的理解和应用。

一、JDK自带的ThreadPoolExecutor

在Java的SDK中，已经帮我们集成好了相应的线程池ThreadPoolExecutor。

先看一下对应的继承树
- 简略图
- UML图
再看看对应的构造方法

如果不想看繁琐的英文，参数含义也可以看下面这张表
以下是线程池核心参数的详细说明表格：

线程池核心参数说明

参数名	含义
corePoolSize	核心线程数，池中常驻的线程数量（包括空闲线程）。当活动线程数小于该值时，直接创建新线程；否则任务进入工作队列等待。
maximumPoolSize	池中允许的最大线程数。当核心线程和工作队列满载且活动线程数小于该值时，创建新线程；超过则触发饱和策略。
keepAliveTime	非核心线程的空闲存活时间。当线程数超过corePoolSize时，空闲线程在指定时间后终止，直到恢复至核心线程数。
unit	keepAliveTime的时间单位（如秒、毫秒）。
workQueue	存储待执行任务的阻塞队列。仅保存通过execute提交的Runnable任务，每个线程从中获取任务顺序执行。
handler	饱和策略。当线程数达到maximumPoolSize且工作队列满载时，处理新任务的策略（如拒绝、丢弃等）。

二、线程池的策略

2.1 常见的任务队列类型

线程池的任务队列用于存储待执行的任务，其类型直接影响线程池的调度行为（如是否创建新线程、任务是否会被拒绝等）。常见队列类型如下：

1. SynchronousQueue（直接提交队列）

特点：不存储任务，仅作为 “传递者”。每次提交任务时，必须有线程立即接收并执行（否则任务会被拒绝，除非线程池能创建新线程）。
适用场景：任务处理速度快、任务数量不稳定（需要灵活创建线程）的场景。例如 CachedThreadPool（缓存线程池）默认使用此队列，配合 maximumPoolSize=Integer.MAX_VALUE，可快速创建线程处理突发任务。
注意：若线程池所有线程都在忙碌，且无法创建新线程（达到 maximumPoolSize），新任务会被拒绝。

2. ArrayBlockingQueue（有界数组队列）

特点：基于数组的有界阻塞队列，必须指定固定容量（如 new ArrayBlockingQueue(100)）。任务按 FIFO 排序。
调度逻辑：当核心线程满后，新任务会进入队列；队列满后，会创建非核心线程（直到达到 maximumPoolSize）；若仍满，则执行拒绝策略。
适用场景：需要控制队列大小（避免内存溢出）、任务量可预估的场景。例如核心线程数固定、任务处理时间稳定的业务。
优点：有界性可避免资源耗尽，适合对系统稳定性要求高的场景。

3. LinkedBlockingQueue（无界 / 有界链表队列）

特点：基于链表的阻塞队列，默认是无界队列（容量为 Integer.MAX_VALUE），也可手动指定容量（有界模式）。任务按 FIFO 排序。
调度逻辑：若为无界队列，当核心线程满后，新任务会一直进入队列，不会创建非核心线程（maximumPoolSize 失效），直到内存溢出。
适用场景：任务处理时间短、任务量稳定（不会暴涨）的场景。例如 FixedThreadPool（固定线程池）默认使用无界的 LinkedBlockingQueue，确保核心线程稳定复用。
注意：无界队列可能因任务堆积导致 OOM，需谨慎使用。

4. PriorityBlockingQueue（优先队列）

特点：无界阻塞队列，任务会按优先级排序（而非 FIFO），需任务实现 Comparable 接口或通过构造器指定比较器。
适用场景：需要按优先级处理任务的场景（如紧急任务优先执行）。例如调度系统中，高优先级任务需先处理。
注意：无界性可能导致 OOM，且优先级排序会增加任务入队 / 出队的开销。

5. DelayedWorkQueue（延迟队列）

特点：ScheduledThreadPoolExecutor（定时任务线程池）的默认队列，任务会在指定延迟时间后才被执行，按延迟时间排序（延迟短的先执行）。
适用场景：定时任务、延迟任务（如定时重试、定时清理）。例如 ScheduledThreadPool 中调度 schedule() 或 scheduleAtFixedRate() 任务。

2.2 常见的拒绝策略

当线程池无法接收新任务时（如线程池已关闭、任务队列满且线程数达 maximumPoolSize），会触发拒绝策略。Java 内置 4 种拒绝策略（均实现 RejectedExecutionHandler 接口）：

1. AbortPolicy（默认策略）

行为：直接抛出 RejectedExecutionException 异常，中断任务提交流程。
适用场景：需要明确感知任务被拒绝的场景（如核心业务，任务丢失会导致严重问题），通过异常可及时报警或处理。
注意：会中断调用者的执行流程，需捕获异常处理。

2. CallerRunsPolicy（调用者运行策略）

行为：让提交任务的线程（调用者线程）自己执行该任务，而非由线程池线程执行。
作用：减缓任务提交速度（调用者线程被占用），给线程池时间处理现有任务，起到 “缓冲” 作用，避免任务丢失。
适用场景：任务量不大、不希望任务丢失的场景（如非核心业务，允许一定延迟）。

3. DiscardPolicy（丢弃策略）

行为：直接丢弃无法处理的新任务，不做任何处理，也不抛出异常。
适用场景：任务无关紧要（如日志收集、统计上报），丢失少量任务不影响系统核心功能。
风险：任务丢失无感知，需谨慎使用。

4. DiscardOldestPolicy（丢弃最旧任务策略）

行为：丢弃任务队列中最旧的未处理任务（队列头部的任务），然后尝试将新任务加入队列。
适用场景：需要处理 “最新任务” 的场景（如实时数据处理），旧任务可被丢弃。
注意：可能会丢弃正在等待的重要任务，需结合业务判断。
- 队列选择：根据任务特性（是否有界、是否需要优先级、处理速度）选择，优先用有界队列（如 ArrayBlockingQueue）避免 OOM。
- 拒绝策略：根据任务重要性选择，核心任务用 AbortPolicy（及时感知），非核心任务用 CallerRunsPolicy 或 DiscardOldestPolicy（避免丢失或缓冲）。实际使用中，也可自定义拒绝策略（实现 RejectedExecutionHandler 接口），例如将任务持久化到数据库或消息队列，后续重试。

三、介绍一下Runnable、Callable和Future

其实提到线程，就不得不提到Runnable、Callable和Future这三者，因为线程里执行的任务类型就是Runnable或Callable，异步返回的结果就是Future。

Runnable 和 Callable 是任务定义接口，前者无返回值 / 不抛 checked 异常，后者有返回值 / 可抛异常。

Future 是结果管理接口，不定义任务，而是用于控制异步任务（获取结果、取消任务等），通常与 Callable 配合使用。

线程池的 execute() 仅接收 Runnable，submit() 可接收两者并返回 Future，是异步编程的核心工具。

特性	`Runnable`	`Callable<V>`	`Future<V>`
接口定义	函数式接口： `@FunctionalInterface` `public interface Runnable {` `void run();` `}`	函数式接口： `@FunctionalInterface` `public interface Callable<V> {` `V call() throws Exception;` `}`	接口： `public interface Future<V> { ... }`（定义异步计算结果）
返回值	无（`void`）	泛型 `V`（返回任务结果）	不直接产生值，通过 `get()` 获取关联 `Callable` 的返回值（或 `null` 若关联 `Runnable`）
异常处理	仅能抛出 `RuntimeException`（`run()` 无 `throws` 声明）	可抛出任意异常（包括受检异常，`call()` 声明 `throws Exception`）	本身不抛出异常，但 `get()` 可能抛出 `ExecutionException`（包装 `Callable` 异常）或 `InterruptedException`
核心作用	定义无结果的任务（如日志打印）	定义有结果的任务（如数据库查询）	管理异步计算结果（获取值、取消任务、判断状态）
是否可取消任务	否（无内置取消机制）	否（通常配合 `Future.cancel()` 实现）	是（提供 `cancel(boolean mayInterruptIfRunning)` 方法）
与线程池配合	通过 `execute(Runnable)` 提交（无返回值）或 `submit(Runnable)`（返回 `Future<?>`，`get()` 为 `null`）	仅能通过 `submit(Callable)` 提交，返回 `Future<V>`	线程池 `submit()` 的返回值，作为获取结果的入口
典型使用场景	无需返回结果的异步操作	需返回结果的异步计算（如并发查询）	配合 `Callable` 管理异步任务（如监控子任务状态、超时取消）

四、手搓简易线程池

4.1 代码

package com.demo.phc.multi_thread
 
import java.util.concurrent.BlockingQueue
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantLock
 
/**
 * 线程池状态
 */
private const val RUNNING = 0    // 运行中
private const val SHUTDOWN = 1   // 关闭中
private const val TERMINATED = 2 // 已终止
 
/**
 * 自定义线程池实现
 * @param corePoolSize 核心线程数
 * @param maximumPoolSize 最大线程数
 * @param keepAliveTime 非核心线程空闲存活时间
 * @param unit 时间单位
 * @param workQueue 任务队列
 * @param rejectHandler 拒绝策略
 */
class ThreadPool(
    private val corePoolSize: Int,
    private val maximumPoolSize: Int,
    private val keepAliveTime: Long,
    private val unit: TimeUnit,
    private val workQueue: BlockingQueue<Runnable> = LinkedBlockingQueue(),
    private val rejectHandler: RejectHandler = DefaultRejectHandler()
) {
    // 线程池状态锁
    private val lock = ReentrantLock()
 
    // 状态条件变量
    private val termination = lock.newCondition()
 
    // 当前状态
    private var state = RUNNING
 
    // 工作线程集合
    private val workers = mutableListOf<Worker>()
 
    init {
        require(corePoolSize >= 0) { "核心线程数不能为负数" }
        require(maximumPoolSize >= corePoolSize) { "最大线程数不能小于核心线程数" }
        require(keepAliveTime >= 0) { "存活时间不能为负数" }
 
        // 初始化核心线程
        for (i in 0 until corePoolSize) {
            addWorker(null)
        }
    }
 
    /**
     * 提交任务到线程池
     */
    fun execute(task: Runnable) {
        if (task == null) throw NullPointerException("任务不能为null")
 
//        println("线程 ${Thread.currentThread().name} 准备获取锁")
        lock.lock()
        try {
//            println("线程 ${Thread.currentThread().name} 成功获取锁，进入临界区")
            // 线程池已关闭，执行拒绝策略
            if (state != RUNNING) {
                rejectHandler.rejectedExecution(task, this)
                return
            }
 
            // 1. 如果当前线程数小于核心线程数，创建新线程执行任务
            if (workers.size < corePoolSize) {
                if (addWorker(task)) return
            }
 
            // 2. 核心线程已满，尝试加入任务队列
            if (workQueue.offer(task)) return
 
            // 3. 队列已满，尝试创建非核心线程
            if (workers.size < maximumPoolSize && addWorker(task)) return
 
            // 4. 所有条件都不满足，执行拒绝策略
            rejectHandler.rejectedExecution(task, this)
        } finally {
            lock.unlock()
//            println("线程 ${Thread.currentThread().name} 释放锁")
        }
    }
 
    /**
     * 添加工作线程
     * @param firstTask 线程首次执行的任务
     * @return 是否添加成功
     */
    private fun addWorker(firstTask: Runnable?): Boolean {
        lock.lock()
        try {
            // 检查线程池状态
            if (state != RUNNING) return false
 
            // 创建工作线程
            val worker = Worker(firstTask)
            val thread = Thread(worker, "ThreadPool-Worker-${workers.size}")
            worker.thread = thread
            workers.add(worker)
            thread.start()
            return true
        } finally {
            lock.unlock()
        }
    }
 
    /**
     * 关闭线程池，不再接受新任务，但会执行完已提交的任务
     */
    fun shutdown() {
        lock.lock()
        try {
            if (state != RUNNING) return
 
            state = SHUTDOWN
            // 中断空闲线程
            interruptIdleWorkers()
        } finally {
            lock.unlock()
        }
    }
 
    /**
     * 立即关闭线程池，尝试中断所有线程，未执行的任务将被丢弃
     */
    fun shutdownNow(): List<Runnable> {
        lock.lock()
        try {
            if (state >= SHUTDOWN) return emptyList()
 
            state = TERMINATED
            // 中断所有线程
            interruptAllWorkers()
 
            // 收集未执行的任务
            val remainingTasks = mutableListOf<Runnable>()
            workQueue.drainTo(remainingTasks)
            return remainingTasks
        } finally {
            lock.unlock()
        }
    }
 
    /**
     * 等待线程池终止
     */
    fun awaitTermination(timeout: Long, unit: TimeUnit): Boolean {
        var nanos = unit.toNanos(timeout)
        lock.lock()
        try {
            while (state != TERMINATED) {
                if (nanos <= 0) return false
                nanos = termination.awaitNanos(nanos)
            }
            return true
        } finally {
            lock.unlock()
        }
    }
 
    /**
     * 中断所有空闲线程
     */
    private fun interruptIdleWorkers() {
        workers.forEach { it.interruptIfIdle() }
    }
 
    /**
     * 中断所有线程
     */
    private fun interruptAllWorkers() {
        workers.forEach { it.interrupt() }
    }
 
    /**
     * 工作线程类
     */
    private inner class Worker(private var firstTask: Runnable?) : Runnable {
        var thread: Thread? = null
        private var isIdle = false
 
        override fun run() {
            var currentTask: Runnable? = firstTask
            try {
                // 循环获取任务执行
                while (currentTask != null || getTask()?.also { currentTask = it } != null) {
                    try {
                        isIdle = false
                        currentTask?.run()
                    } finally {
                        currentTask = null
                        isIdle = true
                    }
                }
            } finally {
                // 线程退出时清理
                processWorkerExit(this)
            }
        }
 
        /**
         * 从队列获取任务
         */
        private fun getTask(): Runnable? {
            var timedOut = false
 
            while (true) {
                lock.lock()
                try {
                    // 线程池已关闭且队列为空，返回null让线程退出
                    if (state >= SHUTDOWN && workQueue.isEmpty()) {
                        return null
                    }
 
                    // 判断是否需要超时等待（非核心线程或已超时）
                    val needTimeout = workers.size > corePoolSize || timedOut
 
                    // 释放锁后再执行阻塞操作，避免锁被长期占用
                    if (needTimeout) {
                        // 非核心线程：超时等待任务
                        lock.unlock() // 先释放锁
                        val task = workQueue.poll(keepAliveTime, unit) // 阻塞等待（无锁）
                        lock.lock() // 重新获取锁
                        if (task == null) {
                            timedOut = true
                            continue // 超时未获取到任务，继续循环检查
                        }
                        return task
                    } else {
                        // 核心线程：一直等待任务
                        lock.unlock() // 先释放锁
                        val task = workQueue.take() // 阻塞等待（无锁）
                        lock.lock() // 重新获取锁
                        return task
                    }
                } catch (e: InterruptedException) {
                    // 线程被中断时，重新检查状态
                    lock.lock()
                    try {
                        if (state >= SHUTDOWN) {
                            return null // 线程池已关闭，退出
                        }
                    } finally {
                        lock.unlock()
                    }
                } finally {
                    // 确保锁最终被释放（防止异常导致锁泄漏）
                    if (lock.isHeldByCurrentThread) {
                        lock.unlock()
                    }
                }
            }
        }
 
        /**
         * 中断空闲线程
         */
        fun interruptIfIdle() {
            if (isIdle) {
                thread?.interrupt()
            }
        }
 
        /**
         * 中断线程
         */
        fun interrupt() {
            thread?.interrupt()
        }
    }
 
    /**
     * 处理工作线程退出
     */
    private fun processWorkerExit(worker: Worker) {
        lock.lock()
        try {
            workers.remove(worker)
 
            // 如果线程池已关闭且没有工作线程了，标记为已终止
            if (state >= SHUTDOWN && workers.isEmpty()) {
                state = TERMINATED
                termination.signalAll()
            }
            // 如果线程池仍在运行且线程数小于核心线程数，补充新线程
            else if (state == RUNNING && workers.size < corePoolSize) {
                addWorker(null)
            }
        } finally {
            lock.unlock()
        }
    }
 
    /**
     * 获取当前线程数
     */
    fun getPoolSize(): Int {
        lock.lock()
        return try {
            workers.size
        } finally {
            lock.unlock()
        }
    }
 
    /**
     * 获取队列中的任务数
     */
    fun getQueueSize(): Int = workQueue.size
 
    /**
     * 拒绝策略接口
     */
    interface RejectHandler {
        fun rejectedExecution(task: Runnable, threadPool: ThreadPool)
    }
 
    /**
     * 默认拒绝策略：抛出异常
     */
    class DefaultRejectHandler : RejectHandler {
        override fun rejectedExecution(task: Runnable, threadPool: ThreadPool) {
            throw RejectedExecutionException("任务被拒绝，线程池已达最大容量")
        }
    }
 
    /**
     * 丢弃任务的拒绝策略
     */
    class DiscardPolicy : RejectHandler {
        override fun rejectedExecution(task: Runnable, threadPool: ThreadPool) {
            // 什么也不做，直接丢弃任务
        }
    }
 
    /**
     * 由调用者线程执行任务的拒绝策略
     */
    class CallerRunsPolicy : RejectHandler {
        override fun rejectedExecution(task: Runnable, threadPool: ThreadPool) {
            if (threadPool.state == RUNNING) {
                // 由提交任务的线程执行
                task.run()
            }
        }
    }
}
 
/**
 * 任务拒绝异常
 */
class RejectedExecutionException(message: String) : RuntimeException(message)
 
// 测试代码
fun main() {
    // 创建线程池：核心2，最大4，队列容量2，非核心线程存活3秒
    val threadPool = ThreadPool(
        corePoolSize = 2,
        maximumPoolSize = 4,
        keepAliveTime = 3,
        unit = TimeUnit.SECONDS,
        workQueue = LinkedBlockingQueue(2),
        rejectHandler = ThreadPool.CallerRunsPolicy()
    )
 
    // 提交8个任务
    for (i in 1..8) {
        try {
            threadPool.execute {
                println("任务$i 开始执行，线程: ${Thread.currentThread().name}")
                Thread.sleep(1000)
                println("任务$i 执行完毕")
            }
        } catch (e: RejectedExecutionException) {
            println("任务$i 被拒绝：${e.message}")
        }
    }
 
    println("当前线程数：${threadPool.getPoolSize()}")
    println("队列任务数：${threadPool.getQueueSize()}")
 
    // 等待任务执行
    Thread.sleep(2000)
    println("2秒后，当前线程数：${threadPool.getPoolSize()}")
 
    // 关闭线程池
    threadPool.shutdown()
    // 等待线程池终止
    threadPool.awaitTermination(5, TimeUnit.SECONDS)
    println("线程池已终止")
}

Penghc_Blog

欢迎来到我的世界～