graph TB
    subgraph "接入层 - Starter"
        Nacos[NacosCloudRefresherHandler]
        Apollo[ApolloRefresherHandler]
        Common[CommonAutoConfiguration]
        Web[WebThreadPoolAdapter]
        Dashboard[DashboardDevController]
    end
    
    subgraph "核心层 - Core"
        Executor[OneThreadExecutor]
        Registry[OneThreadRegistry]
        Monitor[ThreadPoolMonitor]
        Alarm[ThreadPoolAlarmChecker]
        Builder[ThreadPoolExecutorBuilder]
        Queue[ResizableCapacityLinkedBlockingQueue]
        Parser[ConfigParserHandler]
        Notifier[NotifierDispatcher]
    end
    
    subgraph "基础层 - Spring Base"
        Banner[OneThreadBannerHandler]
        Context[ApplicationContextHolder]
    end
    
    subgraph "外部依赖"
        ConfigCenter[配置中心<br/>Nacos/Apollo]
        Metrics[Prometheus<br/>Grafana]
        Notify[钉钉/飞书]
    end
    
    Nacos --> ConfigCenter
    Apollo --> ConfigCenter
    Nacos --> Common
    Apollo --> Common
    
    Common --> Registry
    Common --> Monitor
    Common --> Alarm
    
    Dashboard --> Registry
    Dashboard --> Monitor
    
    Web --> Executor
    
    Registry --> Executor
    Registry --> Holder
    
    Monitor --> Metrics
    Alarm --> Notifier
    Notifier --> Notify
    
    Builder --> Executor
    Builder --> Queue
    
    Executor --> Queue
    Executor --> Holder

// 位置: core/executor/ThreadPoolExecutorHolder.java
@Data
public class ThreadPoolExecutorHolder {
    private String threadPoolId;              // 身份证号
    private ThreadPoolExecutor executor;      // 真身
    private ThreadPoolExecutorProperties executorProperties; // 配置档案
}

// 位置: core/config/BootstrapConfigProperties.java
public class BootstrapConfigProperties {
    private Boolean enable;                   // 是否生效
    private NacosConfig nacos;                // Nacos 配置
    private ConfigFileTypeEnum configFileType; // 配置格式
    private NotifyPlatformsConfig notifyPlatforms; // 通知渠道
    private MonitorConfig monitor;            // 监控配置
    private List<ThreadPoolExecutorProperties> executors; // 所有线程池配置
}

// 位置: core/executor/support/ResizableCapacityLinkedBlockingQueue.java
public class ResizableCapacityLinkedBlockingQueue<E> {
    private volatile int capacity;            // 动态容量（关键！）
    private final AtomicInteger count;        // 当前元素数
    private final ReentrantLock takeLock;     // 取锁
    private final ReentrantLock putLock;      // 放锁
}

NotifierDispatcherResizableCapacityLinkedBlockingQueueOneThreadExecutorOneThreadRegistryDynamicThreadPoolRefreshListenerThreadPoolConfigUpdateEventConfigParserHandlerNacosCloudRefresherHandlerNacos配置中心用户/Nacos控制台NotifierDispatcherResizableCapacityLinkedBlockingQueueOneThreadExecutorOneThreadRegistryDynamicThreadPoolRefreshListenerThreadPoolConfigUpdateEventConfigParserHandlerNacosCloudRefresherHandlerNacos配置中心用户/Nacos控制台alt[参数有变化]修改线程池参数(corePoolSize: 10 → 20)1触发 Listener.receiveConfigInfo()2解析配置内容3返回 Map<Object, Object>4Binder 绑定到 BootstrapConfigProperties5发布 ThreadPoolConfigUpdateEvent6Spring 事件机制触发监听7获取 ThreadPoolExecutorHolder8对比新旧参数9setCorePoolSize(20)10setMaximumPoolSize(30)11setCapacity(500)12更新 executorProperties13发送配置变更通知14调用钉钉 Webhook15

configService.addListener(dataId, group, new Listener() {
    @Override
    public void receiveConfigInfo(String configInfo) {
        refreshThreadPoolProperties(configInfo); // 收到变更立即刷新
    }
});

Map<Object, Object> configInfoMap = ConfigParserHandler.getInstance()
    .parseConfig(configInfo, properties.getConfigFileType());
Binder binder = new Binder(new MapConfigurationPropertySource(configInfoMap));
BootstrapConfigProperties refresherProperties = binder.bind(...);
​
// 发布事件 —— 关键！用事件解耦配置解析和实际刷新
ApplicationContextHolder.getInstance().publishEvent(
    new ThreadPoolConfigUpdateEvent(this, refresherProperties)
);

// 以线程池 ID 为粒度加锁 —— 并发安全关键点！
synchronized (threadPoolId.intern()) {
    boolean changed = hasThreadPoolConfigChanged(remoteProperties);
    if (!changed) continue;
    
    updateThreadPoolFromRemoteConfig(remoteProperties); // 执行参数变更
    sendThreadPoolConfigChangeMessage(...); // 发送通知
}

// 核心线程数变更的特殊处理 —— 先扩容最大线程数，再扩核心数
if (remoteCorePoolSize > originalMaximumPoolSize) {
    executor.setMaximumPoolSize(remoteMaximumPoolSize); // 先扩上限
    executor.setCorePoolSize(remoteCorePoolSize);       // 再设核心
} else {
    executor.setCorePoolSize(remoteCorePoolSize);       // 反之先缩核心
    executor.setMaximumPoolSize(remoteMaximumPoolSize); // 再缩上限
}

// 队列容量动态变更 —— 关键创新点
if (isQueueCapacityChanged(...)) {
    ResizableCapacityLinkedBlockingQueue<?> queue = (ResizableCapacityLinkedBlockingQueue<?>) executor.getQueue();
    queue.setCapacity(remoteProperties.getQueueCapacity()); // 运行时改容量
}

// AbstractDynamicThreadPoolRefresher.java — 模板抽象
public abstract class AbstractDynamicThreadPoolRefresher {
    // 模板方法 —— 定义骨架
    @Override
    public void run(ApplicationArguments args) {
        beforeRegister();      // 钩子1：前置处理
        registerListener();    // 抽象方法：子类实现
        afterRegister();       // 钩子2：后置处理
    }
    
    // 复用逻辑 —— 所有子类共享
    public void refreshThreadPoolProperties(String configInfo) {
        // 解析、绑定、发布事件 — 通用流程
    }
}

// NacosCloudRefresherHandler.java — 具体实现
public void registerListener() {
    configService.addListener(...); // Nacos 特有逻辑
}

// 事件发布
ApplicationContextHolder.getInstance().publishEvent(
    new ThreadPoolConfigUpdateEvent(this, refresherProperties)
);

// 监听器1：业务线程池刷新
public class DynamicThreadPoolRefreshListener 
    implements ApplicationListener<ThreadPoolConfigUpdateEvent> { ... }

// 监听器2：Web 容器线程池刷新
public class WebThreadPoolRefreshListener 
    implements ApplicationListener<ThreadPoolConfigUpdateEvent> { ... }

ThreadPoolExecutor executor = ThreadPoolExecutorBuilder.builder()
    .threadPoolId("onethread-producer")
    .corePoolSize(2)
    .maximumPoolSize(4)
    .keepAliveTime(9999L)
    .workQueueType(BlockingQueueTypeEnum.SYNCHRONOUS_QUEUE)
    .threadFactory("onethread-producer_")
    .rejectedHandler(new ThreadPoolExecutor.CallerRunsPolicy())
    .dynamicPool()  // 标记为动态线程池
    .build();

// OneThreadExecutor.java:138-154
@Override
public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
    RejectedExecutionHandler handlerWrapper = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            rejectCount.incrementAndGet(); // 增强功能
            handler.rejectedExecution(r, executor); // 原有逻辑
        }
    };
    super.setRejectedExecutionHandler(handlerWrapper);
}

if (remoteCorePoolSize > originalMaximumPoolSize) {
    // 扩容场景：先扩上限，再设核心
    executor.setMaximumPoolSize(remoteMaximumPoolSize);
    executor.setCorePoolSize(remoteCorePoolSize);
} else {
    // 缩容场景：先缩核心，再缩上限
    executor.setCorePoolSize(remoteCorePoolSize);
    executor.setMaximumPoolSize(remoteMaximumPoolSize);
}

// ResizableCapacityLinkedBlockingQueue.java:276-283
public void setCapacity(int capacity) {
    final int oldCapacity = this.capacity;
    this.capacity = capacity;
    final int size = count.get();
    if (capacity > size && size >= oldCapacity) {
        signalNotFull(); // 唤醒被阻塞的生产者线程
    }
}

// NotifierDispatcher.java:84-94
boolean allowSend = AlarmRateLimiter.allowAlarm(
    alarm.getThreadPoolId(),
    alarm.getAlarmType(),
    alarm.getInterval()  // 用户自定义间隔（秒）
);
if (allowSend) {
    service.sendAlarmMessage(alarm.resolve());
}

// 在 DynamicThreadPoolConfiguration.java 中新增一个线程池
@Bean
@DynamicThreadPool
public ThreadPoolExecutor myTestPool() {
    return ThreadPoolExecutorBuilder.builder()
        .threadPoolId("my-test-pool")
        .corePoolSize(5)
        .maximumPoolSize(10)
        .workQueueType(BlockingQueueTypeEnum.RESIZABLE_CAPACITY_LINKED_BLOCKING_QUEUE)
        .workQueueCapacity(100)
        .threadFactory("my-test-")
        .dynamicPool()
        .build();
}

@Test
public void testRejectAlarm() {
    ThreadPoolExecutor executor = ThreadPoolExecutorBuilder.builder()
        .threadPoolId("reject-test")
        .corePoolSize(1)
        .maximumPoolSize(1)
        .workQueueCapacity(1)  // 极小队列
        .threadFactory("reject-test-")
        .rejectedHandler(new ThreadPoolExecutor.AbortPolicy())
        .dynamicPool()
        .build();
    
    // 提交超过容量的任务
    for (int i = 0; i < 10; i++) {
        try {
            executor.execute(() -> {
                try { Thread.sleep(10000); } catch (Exception e) {}
            });
        } catch (RejectedExecutionException e) {
            System.out.println("任务被拒绝: " + i);
        }
    }
    
    // 验证拒绝次数
    OneThreadExecutor oneThreadExecutor = (OneThreadExecutor) executor;
    System.out.println("拒绝次数: " + oneThreadExecutor.getRejectCount().get());
}

flowchart LR
    subgraph "运维人员操作"
        A[配置中心控制台<br/>Nacos/Apollo]
    end
    
    subgraph "oneThread 框架"
        B[监听器捕获变更]
        C[解析配置]
        D[对比参数差异]
        E[执行线程池API<br/>setCorePoolSize等]
        F[更新Holder档案]
        G[发送变更通知]
    end
    
    subgraph "线程池实例"
        H[ThreadPoolExecutor<br/>参数实时生效]
    end
    
    subgraph "监控系统"
        I[Prometheus采集]
        J[Grafana可视化]
        K[钉钉告警]
    end
    
    A -->|修改yaml配置| B
    B --> C --> D --> E --> F --> G
    E -->|JDK原生API| H
    H -->|Micrometer指标| I --> J
    H -->|拒绝/队列满| K

远端配置变更 → 监听器感知 → 线程池API调用 → 参数实时生效

onethread:
  executors:
    - thread-pool-id: onethread-producer
      core-pool-size: 20     # 原来是 12
      maximum-pool-size: 30  # 原来是 18

// 注册到 Micrometer，自动采集到 Prometheus
Metrics.gauge("dynamic.thread-pool.active.size", tags, 
    runtimeInfo, ThreadPoolRuntimeInfo::getActivePoolSize);

// 每5秒检查一次
scheduler.scheduleWithFixedDelay(this::checkAlarm, 0, 5, TimeUnit.SECONDS);

private void checkQueueUsage(ThreadPoolExecutorHolder holder) {
    int usageRate = (queueSize * 100) / capacity;
    if (usageRate >= threshold) {
        sendAlarmMessage("Capacity", holder); // 发钉钉消息
    }
}

// 它把"静态的JDK线程池"包装成"动态的智能实体"
public class ThreadPoolExecutorHolder {
    private String threadPoolId;              // 身份标识（Nacos配置的key）
    private ThreadPoolExecutor executor;      // 真正干活的家伙
    private ThreadPoolExecutorProperties executorProperties; // 当前配置档案
}

// JDK ThreadPoolExecutor 提供的运行时修改方法
public void setCorePoolSize(int corePoolSize);      // 修改核心线程数
public void setMaximumPoolSize(int maximumPoolSize); // 修改最大线程数  
public void setKeepAliveTime(long time, TimeUnit unit); // 修改存活时间
public void allowCoreThreadTimeOut(boolean value);   // 核心线程超时设置

// NacosCloudRefresherHandler.java:66-92
public void registerListener() {
    // 向Nacos注册监听器 —— 关键！这是"信号接收器"
    configService.addListener(dataId, group, new Listener() {
        
        @Override
        public void receiveConfigInfo(String configInfo) {
            // Nacos配置变更时，这个方法自动被调用
            refreshThreadPoolProperties(configInfo);
        }
    });
}

// AbstractDynamicThreadPoolRefresher.java:92-96
public void refreshThreadPoolProperties(String configInfo) {
    // 解析yaml/properties配置
    Map<Object, Object> configInfoMap = ConfigParserHandler.getInstance()
        .parseConfig(configInfo, properties.getConfigFileType());
    
    // Spring Boot的Binder将配置绑定到Java对象
    ConfigurationPropertySource sources = new MapConfigurationPropertySource(configInfoMap);
    Binder binder = new Binder(sources);
    BootstrapConfigProperties refresherProperties = binder.bind(...);
    
    // 发布事件，通知所有监听器
    ApplicationContextHolder.getInstance().publishEvent(
        new ThreadPoolConfigUpdateEvent(this, refresherProperties)
    );
}

// DynamicThreadPoolRefreshListener.java:81-123
@Override
public void onApplicationEvent(ThreadPoolConfigUpdateEvent event) {
    // 收到事件后，遍历所有线程池配置
    for (ThreadPoolExecutorProperties remoteProperties : event.getExecutors()) {
        String threadPoolId = remoteProperties.getThreadPoolId();
        
        // 加锁防止并发刷新同一个线程池
        synchronized (threadPoolId.intern()) {
            // 检查参数是否真的变化了
            if (!hasThreadPoolConfigChanged(remoteProperties)) continue;
            
            // 执行真正的参数修改！
            updateThreadPoolFromRemoteConfig(remoteProperties);
        }
    }
}

// DynamicThreadPoolRefreshListener.java:144-178
private void updateThreadPoolFromRemoteConfig(ThreadPoolExecutorProperties remoteProperties) {
    ThreadPoolExecutor executor = holder.getExecutor();
    
    // ===== 核心线程数修改 =====
    Integer remoteCorePoolSize = remoteProperties.getCorePoolSize();
    Integer remoteMaximumPoolSize = remoteProperties.getMaximumPoolSize();
    
    if (remoteCorePoolSize != null && remoteMaximumPoolSize != null) {
        // 关键技巧：扩容时先扩上限，再扩核心；缩容时相反
        if (remoteCorePoolSize > executor.getMaximumPoolSize()) {
            executor.setMaximumPoolSize(remoteMaximumPoolSize); // 先扩上限
            executor.setCorePoolSize(remoteCorePoolSize);       // 再扩核心
        } else {
            executor.setCorePoolSize(remoteCorePoolSize);       // 先缩核心
            executor.setMaximumPoolSize(remoteMaximumPoolSize); // 再缩上限
        }
    }
    
    // ===== 存活时间修改 =====
    if (remoteProperties.getKeepAliveTime() != null) {
        executor.setKeepAliveTime(remoteProperties.getKeepAliveTime(), TimeUnit.SECONDS);
    }
    
    // ===== 拒绝策略修改 =====
    if (remoteProperties.getRejectedHandler() != null) {
        RejectedExecutionHandler handler = RejectedPolicyTypeEnum
            .createPolicy(remoteProperties.getRejectedHandler());
        executor.setRejectedExecutionHandler(handler);
    }
    
    // ===== 队列容量修改（特殊） =====
    if (isQueueCapacityChanged(remoteProperties)) {
        // 获取可变容量队列
        ResizableCapacityLinkedBlockingQueue<?> queue = 
            (ResizableCapacityLinkedBlockingQueue<?>) executor.getQueue();
        // 动态修改队列容量！
        queue.setCapacity(remoteProperties.getQueueCapacity());
    }
}

// java.util.concurrent.ThreadPoolExecutor 原生API
public class ThreadPoolExecutor {
    // 这些方法都是公开的，运行时可直接调用！
    public void setCorePoolSize(int corePoolSize);     // ✓ 动态修改核心线程数
    public void setMaximumPoolSize(int maximumPoolSize); // ✓ 动态修改最大线程数
    public void setKeepAliveTime(long time, TimeUnit unit); // ✓ 动态修改存活时间
    public void allowCoreThreadTimeOut(boolean value);  // ✓ 动态修改核心线程超时
    public void setRejectedExecutionHandler(RejectedExecutionHandler handler); // ✓ 动态修改拒绝策略
}

ThreadPoolExecutor参数刷新器Nacos监听器Nacos配置中心ThreadPoolExecutor参数刷新器Nacos监听器Nacos配置中心第一步：监听器长轮询第二步：解析配置第三步：找出差异第四步：调用JDK API整个过程<1秒完成配置文件变更推送（长连接实时感知）1receiveConfigInfo(configInfo)2调用 refreshThreadPoolProperties()3解析YAML → Map4Binder绑定到对象5对比新旧参数6只处理变化的参数7setCorePoolSize(20)8setMaximumPoolSize(30)9立即生效（无需重启）10

// NacosCloudRefresherHandler.java:66-89
configService.addListener(dataId, group, new Listener() {
    @Override
    public void receiveConfigInfo(String configInfo) {
        // Nacos配置变更时，这个方法会被自动回调
        refreshThreadPoolProperties(configInfo);
    }
});

// AbstractDynamicThreadPoolRefresher.java:92-96
Map<Object, Object> configInfoMap = ConfigParserHandler.getInstance()
    .parseConfig(configInfo, properties.getConfigFileType()); // YAML → Map

ConfigurationPropertySource sources = new MapConfigurationPropertySource(configInfoMap);
Binder binder = new Binder(sources);

// 自动绑定到配置对象（类似 @ConfigurationProperties）
BootstrapConfigProperties refresherProperties = binder.bind(
    "onethread", 
    Bindable.ofInstance(properties)
).get();

private boolean hasDifference(
    ThreadPoolExecutorProperties originalProperties,  // 内存中的旧配置
    ThreadPoolExecutorProperties remoteProperties,    // Nacos的新配置
    ThreadPoolExecutor executor                       // 线程池实例
) {
    // 只检查变化的参数
    return isChanged(originalProperties.getCorePoolSize(), remoteProperties.getCorePoolSize())
        || isChanged(originalProperties.getMaximumPoolSize(), remoteProperties.getMaximumPoolSize())
        || isChanged(originalProperties.getKeepAliveTime(), remoteProperties.getKeepAliveTime())
        || isChanged(originalProperties.getRejectedHandler(), remoteProperties.getRejectedHandler())
        || isQueueCapacityChanged(originalProperties, remoteProperties, executor);
}

private <T> boolean isChanged(T before, T after) {
    return after != null && !Objects.equals(before, after);
}

private void updateThreadPoolFromRemoteConfig(ThreadPoolExecutorProperties remoteProperties) {
    ThreadPoolExecutor executor = holder.getExecutor();
    
    // 直接调用JDK API —— 立即生效！
    
    // 1. 修改核心线程数
    executor.setCorePoolSize(remoteProperties.getCorePoolSize());
    
    // 2. 修改最大线程数
    executor.setMaximumPoolSize(remoteProperties.getMaximumPoolSize());
    
    // 3. 修改存活时间
    executor.setKeepAliveTime(remoteProperties.getKeepAliveTime(), TimeUnit.SECONDS);
    
    // 4. 修改拒绝策略
    RejectedExecutionHandler handler = RejectedPolicyTypeEnum.createPolicy(
        remoteProperties.getRejectedHandler()
    );
    executor.setRejectedExecutionHandler(handler);
    
    // 5. 修改队列容量（oneThread特有）
    if (queue instanceof ResizableCapacityLinkedBlockingQueue) {
        ((ResizableCapacityLinkedBlockingQueue<?>) queue).setCapacity(
            remoteProperties.getQueueCapacity()
        );
    }
}

// JDK ThreadPoolExecutor.java（简化版）
public void setCorePoolSize(int corePoolSize) {
    // 1. 参数校验
    if (corePoolSize < 0 || corePoolSize > maximumPoolSize)
        throw new IllegalArgumentException();
    
    // 2. 记录差值
    int delta = corePoolSize - this.corePoolSize;
    
    // 3. 更新核心线程数
    this.corePoolSize = corePoolSize;
    
    // 4. 如果新核心数比原来大 → 立即创建新线程
    if (delta > 0) {
        // 启动额外的核心线程
        int k = Math.min(delta, workQueue.size());
        while (k-- > 0 && addIfUnderCorePoolSize(null)) {
            // 直接创建新线程！
        }
    }
    
    // 5. 如果新核心数比原来小 → 让多余线程空闲后回收
    else if (delta < 0) {
        // 中断空闲线程，让它们自然退出
        interruptIdleWorkers(-delta);
    }
}

// JDK ThreadPoolExecutor.java
public void setMaximumPoolSize(int maximumPoolSize) {
    // 参数校验
    if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
        throw new IllegalArgumentException();
    
    // 更新最大线程数
    this.maximumPoolSize = maximumPoolSize;
    
    // 如果当前线程数超过新最大值 → 立即中断多余空闲线程
    if (workers.size() > maximumPoolSize) {
        interruptIdleWorkers(workers.size() - maximumPoolSize);
    }
}

// JDK LinkedBlockingQueue.java
public class LinkedBlockingQueue<E> {
    private final int capacity; // final！无法修改
}

// ResizableCapacityLinkedBlockingQueue.java:116
private volatile int capacity; // 改为volatile，可修改！

// 动态修改容量方法
public void setCapacity(int capacity) {
    final int oldCapacity = this.capacity;
    this.capacity = capacity; // 直接赋值
    
    final int size = count.get();
    // 如果扩容且有等待的生产者 → 唤醒它们
    if (capacity > size && size >= oldCapacity) {
        signalNotFull(); // 关键！唤醒被阻塞的put操作
    }
}

时间轴（毫秒级）：
┌────────────────────────────────────────────────────────────────────┐
│ 0ms   │ Nacos控制台：用户修改yaml，点击发布                          │
│       │                                                          │
│ 1ms   │ Nacos Server：检测变更，准备推送                            │
│       │                                                          │
│ 2ms   │ Nacos Client（应用端）：长轮询连接收到推送                   │
│       │                                                          │
│ 3ms   │ Nacos Listener：触发 receiveConfigInfo()                  │
│       │                                                          │
│ 5ms   │ ConfigParser：解析YAML → Map                               │
│       │                                                          │
│ 10ms  │ Binder：绑定到 BootstrapConfigProperties                   │
│       │                                                          │
│ 15ms  │ 对比逻辑：检查哪些参数变化                                  │
│       │                                                          │
│ 20ms  │ 调用 executor.setCorePoolSize()                            │
│       │ → JDK立即创建新线程（如果有任务排队）                        │
│       │                                                          │
│ 25ms  │ 调用 executor.setMaximumPoolSize()                         │
│       │                                                          │
│ 30ms  │ 调用 queue.setCapacity()（如果使用可变队列）                │
│       │ → 原本阻塞的put线程被唤醒继续工作                            │
│       │                                                          │
│ 35ms  │ 发送钉钉通知：参数变更通知                                  │
│       │                                                          │
│ 40ms  │ 完成！整个过程无需重启，线程池已用新参数工作                  │
└────────────────────────────────────────────────────────────────────┘