Java手写线程池感悟与体会

2026-10-11 · 26 分钟 · 分类：折腾记录

线程池是什么

线程池是一种池化技术：预先创建一定数量的线程，并由算法统一管理。有任务产生时直接复用池内的空闲线程，避免频繁创建、销毁线程带来的开销，也避免了OOM（Out Of Memory，内存溢出），从而提升系统的性能与稳定性。

线程池需要用到的组件与知识

并发

多个任务在一段时间内交替执行。由于切换速度极快，宏观上看起来是同时进行，实际上是交替推进。

阻塞队列

一种自带阻塞等待能力的队列。队列为空时调用 take() 会自动阻塞，直到有元素可取；队列满时调用 put() 也会阻塞，直到队列空出位置。

CAS 与 AtomicInteger

CAS（compare and set，比较并交换）是一种无锁的原子操作方法：写入前先比对内存中的值与预期值，一致才写入，否则重试。

AtomicInteger 是 Java 提供的原子整数类，内部就靠 CAS 实现。普通的类在多线程高并发下，如果操作过程没加锁，会因为非原子性出现中间误差，最终结果出错。

线程

程序执行的最小调度单位。Thread 类实现了 Runnable 接口，run() 方法里放需要执行的代码。

启动线程要用 start() 而不是 run()，run() 只是一个普通的方法调用，不会开新线程；start() 内部调用的本地方法 start0() 才真正创建新线程。

锁

高并发下，有些变量需要被多个线程同时访问，这块区域叫临界区，不加保护会导致各种错误。我们希望同一时刻只有一个线程能访问内部资源 —— 锁就是为了实现互斥访问，只有持有锁的线程才能进入临界区。

有了以上知识，就可以开始写代码了

手写线程池

变量区

需要用到的变量如下：

字段	类型	初始值	说明
`coreThreads`	`int`	—	核心线程数（core threads），常驻线程，空闲时也不回收
`maximumThreads`	`int`	—	最大线程数（maximum threads），线程池能开到的上限
`keepAliveTime`	`long`	—	空闲存活时间（keep alive time），线程数超出核心数后，空闲超过这个时长的线程会被回收
`isOn`	`volatile boolean`	`true`	线程池运行开关。`volatile` 保证一个线程改了它，其他线程立刻能看到
`threads`	`List<Thread>`	—	存放工作线程的容器
`runningThreads`	`AtomicInteger`	`0`	当前运行中的线程数。不能用 `threads.size()` 替代 —— `size()` 的访问需要加锁，会更麻烦
`queue`	`BlockingQueue<Runnable>`	—	任务队列，存放等待执行的任务

线程如何工作？

线程采用存活机制：只要活着就持续工作。我们用 while 循环不停地从阻塞队列里取任务。

同时需要区分核心线程和临时线程，但不必严格区分 —— 直接在线程运行前判断当前线程数量，再选用不同的队列方法 poll() / take() 即可。poll() 在超时之后会返回 null，根据这个特性就能判断并销毁临时线程。

线程run()方法代码块

public Runnable MyThreadTask(){
        return ()-> {
            while (isOn) {
                boolean isTmp=runningThreads.get()>coreThreads;
                try {
                    Runnable task;
                    if(isTmp){
                         task= queue.poll(keepAliveTime, TimeUnit.SECONDS);
                    }else{
                        task= queue.take();
                    }
                    if(task==null||task==stop){break;}
                    task.run();
                } catch (InterruptedException e) {
                }
            }
            synchronized (threads) {
                threads.remove(Thread.currentThread());/*threads是链表，因为需
                要经常删改，访问较少（写多读少），所以用链表。
                链表又不支持同时操作，需要上锁*/
            }
            runningThreads.decrementAndGet();
        };
    }

将线程的执行方式写到一个方法里面，方便复用与修改

添加线程

只需根据逻辑判断当前线程数量，然后加减即可。这里用 AtomicInteger 的 compareAndSet() 方法判断是否添加成功。

线程添加代码

compareAndSet(expectValue, newValue) 的实现是：每次操作前先比对内存中的当前值，如果和 expectValue 一致，就更新成功；否则说明值已被别的线程改过，回到循环重新读取、重新尝试。通过这个机制，保证并发时只有一个线程能操作成功。

public boolean addThread(int limit){
    while(true) {
        int c = runningThreads.get();
        if (c >= limit) return false;
        if (runningThreads.compareAndSet(c, c + 1)) {/*失败有两种情况，
        1、值改变但是未超限
        2、值改变同时超限了，
        因此需要重复循环判断*/
            Thread thread = new Thread(MyThreadTask());
            synchronized (threads) {
                threads.add(thread);
            }
            thread.start();
            return true;
        }
    }
}

添加任务

添加新任务的逻辑相对简单，按顺序判断，主要分四种情况：

新任务进来，核心线程未满 → 开核心线程执行（addThread(coreThreads)）
核心线程满了，但任务队列没满 → 任务入队等待（queue.offer）(为什么用offer？因为offer在出现队列已满时不会阻塞，而是返回false)
核心线程满了，队列也满了，但临时线程没满 → 开临时线程执行（addThread(maximumThreads)）
三者都满了 → 拒绝，抛出 RejectedExecutionException，防止任务被静默丢弃

添加任务代码

public void execute(Runnable task) throws InterruptedException{
    if(addThread(coreThreads)) {
        queue.put(task);
    }else if(queue.offer(task)) {
    }else if(addThread(maximumThreads)) {
        queue.put(task);
    }else{
        throw new RejectedExecutionException();
    }
}

关闭线程池

分两种：一种是shutDownNow()，不管当前线程是否执行完毕，立即中断；另一种是shutDown()，等所有任务执行完再关闭线程池。值得一提的是，shutDown()的结束方式是通过循环给队列放入stop这个变量来实现的

shutdownNow() —— 强制关闭

代码块

public List<Runnable> shutdownNow() throws InterruptedException{
    System.out.println("完成剩余任务中~");
    isOn=false;//interrupt可能被线程执行的程序吃掉，isOn保底
    List<Thread> tmp;
    synchronized (threads){
         tmp= new LinkedList<>(threads);
    }
    for(Thread thread:tmp){
        thread.interrupt();
        thread.join();
    }
    List<Runnable> runnables=new ArrayList<>();
    queue.drainTo(runnables);
    System.out.println("线程池关闭完成");
    return runnables;
}

shutdown() —— 优雅关闭

代码块

public void shutdown() throws InterruptedException{
    System.out.println("完成剩余任务中~");
    List<Thread> tmp;
    synchronized (threads) {
        tmp = new LinkedList<>(threads);//防止直接遍历threads导致多线程异常
    }
    //runningThreads.get()直接放到循环内会导致循环的上限一直变化，导致stop的任务数量不够
    for(int i=0;i<tmp.size();i++){
        queue.put(stop);
    }
    for(Thread t : tmp) t.join();
    System.out.println("线程池关闭完成");
}

总结

手写线程池是一个非常好的练习小项目：代码量不大，但能学到很多内容。对并发、线程、CAS、synchronized 等知识点会有更好的理解。由于只是按实际需求简单设计，不是完全照搬 JDK 的实现，所以也没有想象中那么复杂。

踩过的坑

下面这些是开发过程中真实踩到的问题，总代码开头的注释就是这份清单：

#	坑	现象	解法
1	`start()` 写成了 `run()`	线程根本没开起来，变成一次普通方法调用	用 `start()` 启动
2	`count++` 非原子	并发下丢数据	换成 `AtomicInteger`
3	工作内存 vs 主内存	别人改了值，本线程死循环也看不见	变量加 `volatile`
4	`LinkList` 并发增删	抛 CME（ConcurrentModificationException，并发修改异常）	加锁，并且先拷贝再遍历
5	`isOn = false` 了 worker 看不见	线程池关不掉	`isOn` 声明为 `volatile`
6	空转 `while()`	CPU 空转，白白浪费	改用阻塞队列的 `take()` / `poll()`
7	死等的 `shutdown()`	中断被任务代码吞掉，永远等不到	加 `isOn` 标志保底
8	循环上界写成 `runningThreads.get()`	每轮重新计算，stop 的数量对不上	先取 `tmp.size()` 存下来
9	假 CAS：`get` 读两次 / 检查在循环外 / 没重试	并发下判断失效	`while` + `compareAndSet` 合成一体
10	锁漏一半	拷贝 `threads` 时忘了加锁	拷贝也放进 `synchronized` 里

总代码

package org.dromara.MyThreadPool;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

// ── 开发中踩过的 10 个坑（详见正文「踩过的坑」表格）──
//1. start() vs run()               —— 线程没真开
//2. count++ 非原子                  —— 丢数据
//3. 工作内存 vs 主内存 / volatile    —— 死循环看不见
//4. ArrayList 并发增删 → CME        —— 快照 + 锁
//5. isOn=false 了 worker 看不见      —— volatile
//6. 空转 while()                   
//7. 死等的 shutdown                 —— 中断被吞
//8. 循环上界 runningThreads.get()   —— 每轮重算，数量不够
//9. 假 CAS（get读两次/检查在外/没重试）—— 三连坑
//10. 锁漏一半                      —— 拷贝忘加锁（又一次）

public class MyThreadPool{
    int coreThreads;
    int maximumThreads;
    long keepAliveTime;
    volatile boolean isOn=true;
    List<Thread> threads;
    AtomicInteger runningThreads=new AtomicInteger(0);//不可以用list.size来替代，因为size访问的位置都会需要加锁，会导致更麻烦
    BlockingQueue<Runnable> queue;
    private static final Runnable stop = ()->{};
    public MyThreadPool(int coreThreads,int maxThreads,long keepAliveTime,BlockingQueue<Runnable> queue){
        this.coreThreads = coreThreads;
        this.maximumThreads = maxThreads;
        this.keepAliveTime = keepAliveTime;
        threads = new LinkedList<>();
        this.queue=queue;
    }
    public Runnable MyThreadTask(){
        return ()-> {
            while (isOn) {
                boolean isTmp=runningThreads.get()>coreThreads;
                try {
                    Runnable task;
                    if(isTmp){
                         task= queue.poll(keepAliveTime, TimeUnit.SECONDS);
                    }else{
                        task= queue.take();
                    }
                    if(task==null||task==stop){break;}
                    task.run();
                } catch (InterruptedException e) {
                }
            }
            synchronized (threads) {
                threads.remove(Thread.currentThread());
            }
            runningThreads.decrementAndGet();
        };
    }
    public boolean addThread(int limit){
        while(true) {
            int c = runningThreads.get();
            if (c >= limit) return false;
            if (runningThreads.compareAndSet(c, c + 1)) {//失败有两种情况，1、值改变但是未超限2、值改变同时超限了，因此需要重复循环判断
                Thread thread = new Thread(MyThreadTask());
                synchronized (threads) {
                    threads.add(thread);
                }
                thread.start();
                return true;
            }
        }
    }
    public void execute(Runnable task) throws InterruptedException{
        if(addThread(coreThreads)) {
            queue.put(task);
        }else if(queue.offer(task)) {
        }else if(addThread(maximumThreads)) {
            queue.put(task);
        }else{
            throw new RejectedExecutionException();
        }
    }
    public List<Runnable> shutdownNow() throws InterruptedException{
        System.out.println("完成剩余任务中~");
        isOn=false;//interrupt可能被线程执行的程序吃掉，isOn保底
        List<Thread> tmp;
        synchronized (threads){
             tmp= new LinkedList<>(threads);
        }
        for(Thread thread:tmp){
            thread.interrupt();
            thread.join();
        }
        List<Runnable> runnables=new ArrayList<>();
        queue.drainTo(runnables);
        System.out.println("线程池关闭完成");
        return runnables;
    }
    public void shutdown() throws InterruptedException{
        System.out.println("完成剩余任务中~");
        List<Thread> tmp;
        synchronized (threads) {
            tmp = new LinkedList<>(threads);//防止直接遍历threads导致多线程异常
        }
        //runningThreads.get()直接放到循环内会导致循环的上限一直变化，导致stop的任务数量不够
        for(int i=0;i<tmp.size();i++){
            queue.put(stop);
        }
        for(Thread t : tmp) t.join();
        System.out.println("线程池关闭完成");
    }
}

Java 线程池手搓