# dynamic_thread_pool

**Repository Path**: he-wenlin/dynamic_thread_pool

## Basic Information

- **Project Name**: dynamic_thread_pool
- **Description**: 在美团实习期间，通过参照美团的方案设计的动态线程池
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2025-07-08
- **Last Updated**: 2025-08-28

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# dynamic_thread_pool

# 1. 线程池源码介绍

## 1.0. 线程池的状态

​		在ThreadPoolExecutor中，线程池的状态和线程池中线程的数量都是保存在一个ctl变量里面的，高三位为线程池状态，低29位为线程池线程数量。

```java
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 11100000 00000000 00000000 00000000 数值为-536870912
private static final int RUNNING    = -1 << COUNT_BITS;
// 00000000 00000000 00000000 00000000 数值为0
private static final int SHUTDOWN   =  0 << COUNT_BITS;
// 00100000 00000000 00000000 00000000 数值为536870912
private static final int STOP       =  1 << COUNT_BITS;
// 01000000 00000000 00000000 00000000 数值为1073741824
private static final int TIDYING    =  2 << COUNT_BITS;
// 01100000 00000000 00000000 00000000 数值为1610612736
private static final int TERMINATED =  3 << COUNT_BITS;

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
```

线程池状态的数值大小顺序为：*RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED*

- RUNNING：表示线程池在运行中，可以接收新任务，并且可以处理队列中的任务
- SHUTDOWN：不会接收新任务并且会处理队列中的任务
- STOP：不会接收新任务，不会执行队列中的任务，并且会中断正在执行的任务
- TIDYING：所有的任务都终止了，线程池中没有线程了，这时线程池的状态就会变成TIDYING，一旦达到这个状态，就会调用线程池的terminated() 方法
- TERMINATED：终止状态，在TIDYING状态执行完terminated() 方法之后就会转换成这个状态

<img src="./images/线程池状态.png" width="60%">

## 1.1. 线程池核心参数
线程池核心参数包括七个核心参数：
```java
    public ThreadPoolExecutor(int corePoolSize, //核心线程数
                              int maximumPoolSize, // 最大线程数 
                              long keepAliveTime, // 线程存活时间
                              TimeUnit unit, // 线程单位
                              BlockingQueue<Runnable> workQueue, // 工作队列
                              ThreadFactory threadFactory, // 线程工厂
                              RejectedExecutionHandler handler /* 拒绝策略 */) {}
```
线程池的执行流程 execute() 源码如下:
```java
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    // 1. 如果当前线程数量小于核心线程数，那么就新建一个线程去处理任务 command
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true)) // addWorker就是新建线程的逻辑
            return;
        c = ctl.get();
    }
    // 2. 如果核心线程数满了，并且现在状态是RUNNING，那么就将这个任务入队列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 入队后，如果检测到线程池状态不是RUNNING，那么就从队列中移除任务，并执行拒绝策略
        if (! isRunning(recheck) && remove(command)) 
            reject(command);
        // 如果线程数为0，那么就新建线程来处理任务
        else if (workerCountOf(recheck) == 0) // 【比如核心线程全部G了】
            addWorker(null, false);
    }
    // 3. 创建非核心线程处理任务
    else if (!addWorker(command, false))
        // 如果当前线程数大于最大线程数，那么就执行拒绝策略
        reject(command);
}
```
线程池执行流程如 execute() 方法所示：
> 1. 如果当前线程数 < 核心线程数 corePoolSize，那么就会创建核心线程进行处理。
> 2. 如果当前线程数 ≥ 核心线程数 corePoolSize，那么就将任务放在 workQueue 工作队列。
> 3. 如果当前线程数 ≥ 核心线程数 corePoolSize，并且工作队列无法offer元素，那么就创建非核心线程进行处理。
> 4. 如果当前线程数 ≥ 最大线程数 maximumPoolSize，那么就执行拒绝策略 handler。

### 1.1.1.  addWorker() 新建线程的逻辑

```java
// addWorker()的核心逻辑就是让workerCount+1，并且新建一个线程进行任务处理
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        // 检查线程池的状态（线程池可能被其他线程调用shutdown，所以这里要做检查！）
        if (rs >= SHUTDOWN &&
            // 如果线程池状态是SHUTDOWN，并且线程池是非空的话，那么就不会执行这个if
            !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
            return false;

        // 自旋 + cas 将workerCount的数量+1
        for (;;) {
            int wc = workerCountOf(c);
            // 条件判断
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 一次cas将workerCount计数加一，成功就break，否则继续旋转内层循环
            if (compareAndIncrementWorkerCount(c)) 
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    // 上面的代码将 workerCount + 1
    // 下面的代码才是真正新建Worker的步骤
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 新建worker对象
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());
                
                // 如果在 运行中 或者是 SHUTDOWN状态 & 核心线程数为0
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // 将新建的worker添加到 Hashset<Worker>
                    workers.add(w);
                    // 更新ThreadPoolExecutor中的历史最大创建的线程个数
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                        
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 如果worker添加成功，那么就直接开启线程执行任务即可
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // 如果线程开启失败，那么就将workerCount减1，相当于回滚
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}
```

```java
// 开启线程失败后，将workerCount减去1的逻辑
private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (w != null)
            workers.remove(w);
        // 1. 将workerCount减去1的逻辑
        decrementWorkerCount();
        // 2. 线程退出时的逻辑
        tryTerminate();
    } finally {
        mainLock.unlock();
    }
}
```

### 1.1.2.  tryTerminate() 线程退出时的逻辑

```java
final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        // 1. RUNNING状态
        // 2. 已经是TIDYING状态
        // 3. 是SHUWDOWN，但是线程池还有任务
        // 以上几种情况不需要执行 TIDYING 和 TERMINATED 状态切换任务
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
            
        if (workerCountOf(c) != 0) { // 如果workerCount为0，那么就不需要打断
            // 打断一个空闲线程
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        // 1. 最后一个线程将状态设置为 TIDYING
        // 2. 执行完terminated()方法后，将状态设置为TERMINATED
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated(); // 拓展点
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}
```

## 1.2. Worker对象

Worker 利用 AQS 实现了不可重入锁：

```java
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** 第一个任务，可能为空，为空的时候，这个线程就执行线程池中的任务 */
    Runnable firstTask;
    /** Per-thread task counter 每个线程完成的任务数 */
    volatile long completedTasks;

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
        setState(-1); // 初始化AQS状态为-1，目的是禁止中断直到runWorker执行
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        // 线程执行的核心代码
        runWorker(this);
    }

    // 利用AQS实现了一个不可重入的互斥锁，0代表无锁，1代表已经加锁
    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    // 尝试加锁，使用cas将state变成0，如果cas成功，那么就将exclusiveOwnerThread 设置为当前线程
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    // 释放锁，将exclusiveOwnerThread设置为null，state设置为0
    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }
    
    // getState() >= 0：Worker已启动（构造函数中初始化为-1，runWorker时才会置为0）
    // thread != null：线程已成功创建
    // !t.isInterrupted()：线程尚未被中断
    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}
```

### 1.2.1. runWorker() 执行任务

```java
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    // 获取第一个任务
    Runnable task = w.firstTask;
    
    w.firstTask = null;
    w.unlock(); // 允许线程被打断：因为这个Worker创建出来，设置了state为-1，防止被打断
    
    boolean  = true;
    try {
        // 执行第一次提交的任务 or 从BlockingQueue中取任务
        // 跳出while循环，就执行销毁逻辑
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) 
                         && !wt.isInterrupted())
                wt.interrupt();
            
            try {
                // 任务执行前的拓展点
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    // 任务执行后的拓展点
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                // 执行过的任务数量加一
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}
```

### 1.2.2. getTask() 获取队列中的任务

```java
private Runnable getTask() {
    boolean timedOut = false; // 最后一次poll是否超时，在第一次poll后设置为true

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
		
        // 如果线程池状态为 SHUTDOWN&队列为空，或者为STOP状态，那么就不需要获取任务
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            // 执行workerCount减1，返回空之后，线程正常退出。
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        
        // 这里的代码就是销毁线程的代码，直接返回null，没有任务就可以直接执行
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            // 将workerCount减去1，返回空
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            // 这里就是超时时间的逻辑：
                // 如果非核心线程额等待 keepAliveTime时间没有任务，那么就第二轮销毁
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take(); // 会一直阻塞，指导 workQueue 有任务
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}
```

# 2. 动态线程池的介绍

在线程池Worker执行每个任务时，都会执行前置增强`beforeExecute()` 和后置增强`afterExecute()`，在线程池状态从 TIDYING 变成 TERMINATED时，会执行`terminated()` 拓展点。此线程池基于这些拓展点，进行了线程池的一些指标统计和结束工作。

## 2.1. 线程池指标统计

### 2.1.1. 任务执行耗时统计

基于Alibaba开源的`TransmittableThreadLocal`实现线程池任务执行耗时统计，通过线程池的`beforeExecute`和`afterExecute`拓展点进行精准时间测量。该方案解决了传统ThreadLocal在线程池环境下值传递失效的问题，确保任务执行时间的准确统计。

核心实现类`TtlTaskTimeHolder`源码如下：

```java
import com.alibaba.ttl.TransmittableThreadLocal;

/**
 * 基于TransmittableThreadLocal的任务执行时间统计工具
 * 功能特性：
 * 1. 线程安全的开始时间记录
 * 2. 自动计算执行耗时
 * 3. 严格的资源清理机制
 * 4. 支持线程池环境下的值传递
 */
public class TtlTaskTimeHolder {
    // 使用TransmittableThreadLocal记录任务开始时间
    private static final TransmittableThreadLocal<Long> startTimeHolder = new TransmittableThreadLocal<>();
    
    private TtlTaskTimeHolder() {}
    
    /**
     * 记录任务开始时间
     * 调用时机：任务开始执行前
     */
    public static void recordStartTime() {
        startTimeHolder.set(System.currentTimeMillis());
    }
    
    /**
     * 计算并获取任务执行时间
     * @return 执行时间(毫秒)，如果没有记录则返回null
     * 注意：该方法会自动清理ThreadLocal
     */
    public static Long getAndRemoveExecutionTime() {
        Long startTime = startTimeHolder.get();
        if (startTime == null) {
            return null;
        }
        
        long executionTime = System.currentTimeMillis() - startTime;
        startTimeHolder.remove();
        return executionTime;
    }
    
    /**
     * 强制清理ThreadLocal
     * 用于异常情况下的资源释放
     */
    public static void clear() {
        startTimeHolder.remove();
    }
}
```

监控线程池实现类`MonitorThreadPool`源码如下：

```java
import java.util.concurrent.*;

/**
 * 支持任务耗时监控的线程池实现
 * 监控能力：
 * 1. 任务执行耗时统计
 * 2. 超时阈值告警
 * 3. 线程池生命周期管理
 */
public class MonitorThreadPool extends ThreadPoolExecutor {
    // 默认执行时间阈值(毫秒)
    private final long executionThresholdMs;
    
    public MonitorThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                          TimeUnit unit, BlockingQueue<Runnable> workQueue,
                          long executionThresholdMs) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        this.executionThresholdMs = executionThresholdMs;
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        // 记录精确到毫秒的任务开始时间
        TtlTaskTimeHolder.recordStartTime();
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        try {
            // 获取精确执行耗时（自动清理ThreadLocal）
            Long executionTime = TtlTaskTimeHolder.getAndRemoveExecutionTime();
            if (executionTime != null) {
                // 耗时统计与告警处理
                monitorExecutionTime(executionTime);
            }
        } finally {
            // 确保资源释放
            TtlTaskTimeHolder.clear();
            super.afterExecute(r, t);
        }
    }
    
    /**
     * 执行耗时监控核心逻辑
     * @param executionTime 实际执行时间(毫秒)
     */
    private void monitorExecutionTime(long executionTime) {
        // 标准输出格式：[时间类型] 任务耗时
        System.out.printf("[TTL-METRIC] Task execution time: %dms%n", executionTime);
        
        // 超时告警逻辑
        if (executionTime > executionThresholdMs) {
            System.err.printf("[TTL-ALERT] Task execution time %dms exceeds threshold %dms%n",
                            executionTime, executionThresholdMs);
            // 此处可接入告警系统（如邮件、短信等）
        }
    }
}
```

### 2.1.2. 线程池退出时的配置中心信息清理

在`MonitorThreadPool`中重写`terminated()`方法，当线程池完全终止时自动清理Redis注册信息：

```java
public class MonitorThreadPool extends ThreadPoolExecutor {
    /**
     * 线程池终止时自动调用
     */
    @Override
    protected void terminated() {
        try {
            // 清理配置中心的配置信息
            cleanRegistration();
        } finally {
            super.terminated();
        }
    }

    /**
     * 从Redis清理注册信息
     */
    private void cleanRegistration() {
        // todo
    }
}
```

# 3. SPI 服务发现机制

> 可以基于【 Redis的发布订阅模式】和【Zookeeper的watch监听机制】实现动态配置更新。

## 3.1. Redis Pub/Sub 机制

3.1. 基于 Redis List 结构实现动态配置管理

Redis List数据类型中，

更新人群标签？ -> 更新bitmap？bit判断 -> 

## 3.2. Zookeeper 实现动态配置更新

### 3.2.1. zookeeper 命令记录

```shell
# 1. 进入bin文件夹，输入以下命令行，启动客户端[通过2181端口连接的]
./zkCli.sh
# 2. 查看zookeeper提供的常见命令行
help 

# 3. znode节点操作
# 3.1. 查看当前节点有哪些子节点 ls path
ls /
# 3.2. 查看当前节点详细数据 ls -s path
[zk: localhost:2181(CONNECTED) 11] ls -s /
[dubbo, zookeeper]
cZxid = 0x0
ctime = Thu Jan 01 00:00:00 UTC 1970
mZxid = 0x0
mtime = Thu Jan 01 00:00:00 UTC 1970
pZxid = 0x3
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 2

# 3.3. 创建节点 create path
> create /node1
Created /node1
# 3.4. 创建节点，并且保存数据 create path data
create /provider_1 '192.168.5.67'
# 3.5. 获取节点中的数据 get path
get /provider_1
# 3.6. zk无法创建多级路径，必须逐层创建。比如/a/b/c创建过程如下：
create /a
create /a/b
create /a/b/c

# 4. 创建临时节点和顺序节点
# create [-s] [-e] [-c] [-t ttl] <path> [data] [acl]
# -s【强调顺序性】:sequential 顺序节点，创建的节点名字会附加一个单调递增的数字，eg: /node0000000001
	# 用途：用于实现分布式锁、队列等需要有序节点的场景。
	# 示例：create -s /tasks/task_  # 实际路径可能是 `/tasks/task_0000000001`
# -e【强调临时性】:ephemeral 临时节点，节点的生命周期和客户端会话绑定，会话结束就将节点自动删除
	# 用途：临时服务注册（如 Dubbo 的 Provider 节点）。
	# 示例：create -e /services/provider1 "192.168.1.1:8080"
# -c【强调可自动清理的】:container 容器节点，特殊类型的节点，当没有子节点时可能被自动删除（需zk 3.5+）
	# 用途：优化存储空间，自动清理空节点（需配置支持）。
	# create -c /containers/my_container
# -t ttl	Time-To-Live	节点存活时间（毫秒）
# [acl]	Access Control List	访问控制列表

# 5. 修改节点中的数据 set path data
set /a '新数据'
# 6. 删除一个节点
# 6.1 只删除子节点
delete /a/b
# 6.2. 递归删除（删除指定节点和该节点下所有子节点）
deleteall /a # 删除a和a下所有的子节点
```

### 3.2.2. 节点 - 事件监听watch

```shell
# addWatch命令可以让客户端监听指定节点的值变化 或者 子节点的变化（路径发生变化）

# 操作1：客户端1执行 addWatch /a，监听/a节点
# 操作2：客户端2执行 set /a '新内容'
# 可以查看客户端1监听到/a的输出：
WatchedEvent state:SyncConnected type:NodeDataChanged path:/a

# 操作3：客户端1执行 addWatch /a，监听/a节点
# 操作4：客户端2执行 create /a/b 'hah'
# 可以查看客户端1监听到/a的输出：
WatchedEvent state:SyncConnected type:NodeCreated path:/a/b
```