性能优化 ⭐

public class UserService {
 
    // ❌ 反面示例：整个方法加锁，锁范围过大
    public synchronized void updateUser(User user) {
        // ---------- 以下是无需同步的逻辑 ----------
        validateParams(user);           // 参数校验，纯计算，无共享状态
        String formatted = formatName(user.getName()); // 字符串格式化
        logRequest(user);               // 打印日志
 
        // ---------- 以下才是真正需要同步的逻辑 ----------
        db.update(user);                // 写入数据库（共享资源）
        cache.invalidate(user.getId()); // 清除缓存（共享资源）
    }
}

public class UserService {
 
    // 专门用于保护数据库 + 缓存操作的锁对象
    private final Object dbLock = new Object();
 
    // ✅ 正面示例：只锁真正需要同步的临界区
    public void updateUser(User user) {
        // ---------- 无需同步的逻辑，多线程可并行执行 ----------
        validateParams(user);           // 参数校验，线程安全
        String formatted = formatName(user.getName()); // 纯函数，线程安全
        logRequest(user);               // 日志打印，线程安全
 
        // ---------- 仅对共享资源的操作加锁 ----------
        synchronized (dbLock) {         // 临界区开始
            db.update(user);            // 写入数据库
            cache.invalidate(user.getId()); // 清除缓存
        }                               // 临界区结束，锁立即释放
    }
}

public class CoarseGrainedMap<K, V> {
 
    private final Map<K, V> map = new HashMap<>(); // 底层数据存储
    private final Object lock = new Object();       // 唯一的一把大锁
 
    public void put(K key, V value) {
        synchronized (lock) {        // 所有写操作竞争同一把锁
            map.put(key, value);
        }
    }
 
    public V get(K key) {
        synchronized (lock) {        // 所有读操作也竞争同一把锁
            return map.get(key);
        }
    }
}

public class StripedMap<K, V> {
 
    // 分段数量，通常设为 2 的幂次以便位运算取模
    private static final int SEGMENT_COUNT = 16;
 
    // 每个段是一个独立的 HashMap
    private final Map<K, V>[] segments;
 
    // 每个段拥有独立的锁对象
    private final Object[] locks;
 
    @SuppressWarnings("unchecked")
    public StripedMap() {
        segments = new HashMap[SEGMENT_COUNT];  // 创建 16 个段
        locks = new Object[SEGMENT_COUNT];      // 创建 16 把锁
        for (int i = 0; i < SEGMENT_COUNT; i++) {
            segments[i] = new HashMap<>();      // 初始化每个段
            locks[i] = new Object();            // 初始化每把锁
        }
    }
 
    // 根据 key 的哈希值，定位到对应的段索引
    private int segmentIndex(K key) {
        // 取绝对值后对段数取模，确保索引在 [0, SEGMENT_COUNT) 范围内
        return (key.hashCode() & 0x7FFFFFFF) % SEGMENT_COUNT;
    }
 
    public void put(K key, V value) {
        int idx = segmentIndex(key);         // 1. 定位段索引
        synchronized (locks[idx]) {          // 2. 只锁该段的锁
            segments[idx].put(key, value);   // 3. 操作该段的 HashMap
        }
    }
 
    public V get(K key) {
        int idx = segmentIndex(key);         // 1. 定位段索引
        synchronized (locks[idx]) {          // 2. 只锁该段的锁
            return segments[idx].get(key);   // 3. 从该段读取
        }
    }
}

public class OrderService {
 
    private final Object lock = new Object();
    private final List<Order> orders = new ArrayList<>();
 
    // ❌ 反面示例：在锁内构建对象、做格式化
    public void placeOrderBad(String rawData) {
        synchronized (lock) {
            // 以下三步都在锁内完成，但前两步完全不依赖共享状态
            Order order = parseOrder(rawData);     // JSON 解析，耗时 ~5ms
            order.setTimestamp(formatTime());       // 时间格式化，耗时 ~1ms
            orders.add(order);                      // 真正需要同步的：写入共享列表
        }
    }
 
    // ✅ 正面示例：预计算，只把最终赋值放进锁里
    public void placeOrderGood(String rawData) {
        // ---------- 锁外预计算：这些操作线程安全，无需同步 ----------
        Order order = parseOrder(rawData);     // JSON 解析，锁外完成
        order.setTimestamp(formatTime());       // 时间格式化，锁外完成
 
        // ---------- 临界区极短：只做一次 List.add ----------
        synchronized (lock) {
            orders.add(order);                 // 仅此一行需要同步
        }
    }
}

// ❌ 临界区内使用 LinkedList 遍历查找 → O(n)
synchronized (lock) {
    for (Node node : linkedList) {       // 逐个遍历
        if (node.getId() == targetId) {  // 匹配目标
            return node;
        }
    }
}
 
// ✅ 临界区内使用 HashMap 直接定位 → O(1)
synchronized (lock) {
    return hashMap.get(targetId);        // 常量时间查找
}

public void transferMoney(Account from, Account to, double amount) {
    TransferRecord record;
 
    // ---------- 临界区：只做核心的状态变更 ----------
    synchronized (lock) {
        from.debit(amount);              // 扣款
        to.credit(amount);               // 入账
        record = new TransferRecord(from, to, amount); // 构建记录
    }
    // ---------- 锁外完成非关键的后续操作 ----------
 
    auditLog.append(record);             // 写审计日志（可异步）
    notificationService.send(record);    // 发送通知（可异步）
}

// ❌ 所有操作都互斥，读读也互斥
public synchronized String getConfig(String key) {
    return configMap.get(key);       // 读操作
}
 
public synchronized void setConfig(String key, String value) {
    configMap.put(key, value);       // 写操作
}

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
 
public class ConfigCenter {
 
    // 底层存储：配置项的键值对
    private final Map<String, String> configMap = new HashMap<>();
 
    // 读写锁实例，ReentrantReadWriteLock 是 ReadWriteLock 的标准实现
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
 
    /**
     * 读取配置 —— 使用读锁
     * 多个线程可以同时持有读锁，并行读取
     */
    public String getConfig(String key) {
        rwLock.readLock().lock();          // 获取读锁（共享锁）
        try {
            return configMap.get(key);     // 读取共享数据
        } finally {
            rwLock.readLock().unlock();    // 确保释放读锁，防止死锁
        }
    }
 
    /**
     * 批量读取 —— 同样使用读锁，可以和 getConfig 并行
     */
    public Map<String, String> getAllConfigs() {
        rwLock.readLock().lock();          // 获取读锁
        try {
            // 返回快照副本，避免外部修改影响内部数据
            return new HashMap<>(configMap);
        } finally {
            rwLock.readLock().unlock();    // 释放读锁
        }
    }
 
    /**
     * 写入/更新配置 —— 使用写锁
     * 写锁是排他的，获取写锁时必须等待所有读锁和写锁释放
     */
    public void setConfig(String key, String value) {
        rwLock.writeLock().lock();         // 获取写锁（排他锁）
        try {
            configMap.put(key, value);     // 修改共享数据
        } finally {
            rwLock.writeLock().unlock();   // 释放写锁
        }
    }
 
    /**
     * 删除配置 —— 使用写锁
     */
    public void removeConfig(String key) {
        rwLock.writeLock().lock();         // 获取写锁
        try {
            configMap.remove(key);         // 删除共享数据
        } finally {
            rwLock.writeLock().unlock();   // 释放写锁
        }
    }
}

public String computeAndCache(String key) {
    rwLock.readLock().lock();                     // 1. 先获取读锁检查缓存
    try {
        String cached = configMap.get(key);
        if (cached != null) {
            return cached;                         // 缓存命中，直接返回
        }
    } finally {
        rwLock.readLock().unlock();               // 2. 释放读锁
    }
 
    // 缓存未命中，需要写入
    rwLock.writeLock().lock();                     // 3. 获取写锁
    try {
        // Double-check：防止在等待写锁期间其他线程已经写入
        String cached = configMap.get(key);
        if (cached != null) {
            return cached;
        }
 
        String computed = expensiveCompute(key);   // 耗时计算
        configMap.put(key, computed);              // 写入缓存
 
        // ====== 锁降级开始 ======
        rwLock.readLock().lock();                  // 4. 在持有写锁的同时获取读锁
        // ====== 锁降级完成 ======
 
        return computed;
    } finally {
        rwLock.writeLock().unlock();               // 5. 释放写锁（此时仍持有读锁）
        // 此后只持有读锁，其他读线程可以并行进入
        rwLock.readLock().unlock();                // 6. 最终释放读锁
    }
}

import java.util.concurrent.locks.StampedLock;
 
public class Point {
    private double x, y;                              // 共享可变坐标
    private final StampedLock sl = new StampedLock();  // StampedLock 实例
 
    // 写操作：与 ReadWriteLock 类似，获取排他写锁
    public void move(double deltaX, double deltaY) {
        long stamp = sl.writeLock();                   // 获取写锁，返回一个 stamp 戳记
        try {
            x += deltaX;                               // 修改 x 坐标
            y += deltaY;                               // 修改 y 坐标
        } finally {
            sl.unlockWrite(stamp);                     // 使用同一个 stamp 释放写锁
        }
    }
 
    // 乐观读：无锁读取，极致性能
    public double distanceFromOrigin() {
        long stamp = sl.tryOptimisticRead();           // 获取乐观读戳记（不加锁！）
        double currentX = x;                           // 读取 x（可能被并发修改）
        double currentY = y;                           // 读取 y（可能被并发修改）
 
        if (!sl.validate(stamp)) {                     // 验证：读取期间是否有写操作发生？
            // 乐观读失败，回退为悲观读锁
            stamp = sl.readLock();                     // 获取传统读锁
            try {
                currentX = x;                          // 重新读取 x
                currentY = y;                          // 重新读取 y
            } finally {
                sl.unlockRead(stamp);                  // 释放读锁
            }
        }
 
        return Math.sqrt(currentX * currentX + currentY * currentY); // 计算距离
    }
}

// CAS 伪代码 —— 理解其语义
boolean compareAndSwap(内存地址 V, 期望值 A, 新值 B) {
    // ---- 以下整个过程是一条 CPU 指令，不可分割 ----
    if (V 当前的值 == A) {   // Compare：当前值是否等于期望值？
        V 的值 = B;          // Swap：是的话，写入新值
        return true;         // 告诉调用者：成功了
    }
    return false;            // 当前值已被别人改过，失败
}

public class AtomicInteger extends Number implements java.io.Serializable {
 
    // Unsafe 实例 —— 提供 CAS 的底层能力
    private static final Unsafe unsafe = Unsafe.getUnsafe();
 
    // value 字段在对象内存中的偏移量（offset）
    // CAS 需要知道"改哪个内存地址"，这个 offset 就是定位依据
    private static final long valueOffset;
 
    static {
        try {
            // 通过反射获取 "value" 字段的内存偏移
            valueOffset = unsafe.objectFieldOffset(
                AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }
 
    // volatile 保证可见性：任何线程修改后，其他线程立即可见
    private volatile int value;
 
    // ===== 核心方法：getAndIncrement（等价于 i++，但是原子的）=====
    public final int getAndIncrement() {
        return unsafe.getAndAddInt(this, valueOffset, 1);
    }
}

// Unsafe.getAndAddInt 的等价逻辑
public final int getAndAddInt(Object obj, long offset, int delta) {
    int oldValue;                              // 存放"我认为的当前值"
    do {
        oldValue = getIntVolatile(obj, offset); // 第1步：读取当前最新值（volatile 读）
    } while (!compareAndSwapInt(                // 第2步：CAS 尝试写入
            obj,                                //   → 目标对象
            offset,                             //   → 字段偏移量
            oldValue,                           //   → 期望值 = 刚才读到的值
            oldValue + delta));                 //   → 新值 = 旧值 + delta
    // 如果 CAS 失败（说明有别的线程改过了），循环重试
    return oldValue;                            // 返回操作前的旧值
}

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.CountDownLatch;
 
public class AtomicCounterDemo {
 
    // 使用 AtomicInteger 替代 int + synchronized
    private static final AtomicInteger counter = new AtomicInteger(0);
 
    public static void main(String[] args) throws InterruptedException {
        int threadCount = 100;                          // 100 个线程
        int incrementPerThread = 10_000;                // 每个线程自增 10000 次
        CountDownLatch latch = new CountDownLatch(threadCount); // 等待所有线程完成
 
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                for (int j = 0; j < incrementPerThread; j++) {
                    counter.incrementAndGet();           // 原子自增，无需加锁
                }
                latch.countDown();                       // 当前线程完成，计数减 1
            }).start();
        }
 
        latch.await();                                   // 阻塞主线程直到所有子线程完成
        // 结果一定是 1,000,000 —— 不会有线程安全问题
        System.out.println("Final count: " + counter.get());
    }
}

时间线：
T1:  读到 A ────────────────────────── CAS(A → C) ✅ 成功（但中间发生了变化！）
T2:       A → B → A（偷偷改了一圈又改回来）

import java.util.concurrent.atomic.AtomicStampedReference;
 
public class ABADemo {
    public static void main(String[] args) {
        // 初始值 = "A"，初始版本号 stamp = 1
        AtomicStampedReference<String> ref =
                new AtomicStampedReference<>("A", 1);
 
        int[] stampHolder = new int[1];                      // 用数组接收当前 stamp
        String current = ref.get(stampHolder);               // 读取当前值和版本号
        int currentStamp = stampHolder[0];                   // 当前版本号 = 1
 
        // CAS 时不仅比较值，还要比较版本号
        boolean success = ref.compareAndSet(
                current,              // 期望引用值 = "A"
                "C",                  // 新引用值 = "C"
                currentStamp,         // 期望版本号 = 1
                currentStamp + 1);    // 新版本号 = 2
 
        System.out.println("CAS result: " + success);       // true
        // 即使另一个线程把值改成 B 又改回 A，版本号也已经变了（1→2→3）
        // 所以用旧的 stamp=1 去 CAS 会失败 —— ABA 问题被解决
    }
}

import java.util.concurrent.atomic.LongAdder;
 
public class LongAdderDemo {
 
    private static final LongAdder adder = new LongAdder(); // 替代 AtomicLong
 
    public static void main(String[] args) throws Exception {
        int threads = 100;
        int loops = 1_000_000;
        Thread[] pool = new Thread[threads];
 
        for (int i = 0; i < threads; i++) {
            pool[i] = new Thread(() -> {
                for (int j = 0; j < loops; j++) {
                    adder.increment();  // 内部分散到不同 Cell，减少 CAS 冲突
                }
            });
            pool[i].start();
        }
 
        for (Thread t : pool) t.join();  // 等待所有线程完成
 
        // sum() 遍历 base + 所有 Cell 的值求和
        System.out.println("Total: " + adder.sum()); // 100,000,000
    }
}

// ❌ 方案一：共享实例 —— 线程不安全！
public class UnsafeFormatter {
    // SimpleDateFormat 内部有可变状态（calendar 字段），多线程共享会出错
    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
 
    public static String format(Date date) {
        return sdf.format(date);  // 💥 多线程下可能返回错误结果
    }
}
 
// ⚠️ 方案二：每次 new —— 安全但浪费
public class ExpensiveFormatter {
    public static String format(Date date) {
        // 每次调用都创建新对象，GC 压力大
        return new SimpleDateFormat("yyyy-MM-dd").format(date);
    }
}
 
// ✅ 方案三：ThreadLocal —— 安全且高效
public class ThreadLocalFormatter {
    // 每个线程第一次 get() 时调用 initialValue() 创建自己的实例
    private static final ThreadLocal<SimpleDateFormat> formatter =
            ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd"));
 
    public static String format(Date date) {
        return formatter.get().format(date);  // 拿的是当前线程自己的副本
    }
}

// Thread.java 中的关键字段
public class Thread implements Runnable {
    // 每个线程实例都有一个专属的 ThreadLocalMap
    ThreadLocal.ThreadLocalMap threadLocals = null;
}

┌───────────────────────────────────────────────────────────────────┐
│  Thread-1                                                         │
│  ┌─────────────────────────────────────────────────────────────┐  │
│  │  ThreadLocalMap                                             │  │
│  │  ┌────────────┬────────────┬────────────┬───────────────┐  │  │
│  │  │  Entry[0]  │  Entry[1]  │  Entry[2]  │     ...       │  │  │
│  │  │ Key: TL_A  │ Key: TL_B  │   null     │               │  │  │
│  │  │ Val: "abc" │ Val: sdf   │            │               │  │  │
│  │  └────────────┴────────────┴────────────┴───────────────┘  │  │
│  └─────────────────────────────────────────────────────────────┘  │
│                                                                   │
│  Thread-2                                                         │
│  ┌─────────────────────────────────────────────────────────────┐  │
│  │  ThreadLocalMap                                             │  │
│  │  ┌────────────┬────────────┬────────────┬───────────────┐  │  │
│  │  │  Entry[0]  │  Entry[1]  │  Entry[2]  │     ...       │  │  │
│  │  │ Key: TL_A  │   null     │ Key: TL_C  │               │  │  │
│  │  │ Val: "xyz" │            │ Val: conn   │               │  │  │
│  │  └────────────┴────────────┴────────────┴───────────────┘  │  │
│  └─────────────────────────────────────────────────────────────┘  │
└───────────────────────────────────────────────────────────────────┘

// ThreadLocal.get() 简化源码
public T get() {
    Thread t = Thread.currentThread();       // 第1步：获取当前线程
    ThreadLocalMap map = t.threadLocals;     // 第2步：拿到该线程的 ThreadLocalMap
    if (map != null) {
        // 第3步：以"当前 ThreadLocal 实例"为 Key 查找 Entry
        ThreadLocalMap.Entry e = map.getEntry(this);
        if (e != null) {
            return (T) e.value;              // 找到了，直接返回
        }
    }
    return setInitialValue();                // 第4步：没找到，调用 initialValue() 创建
}

// ThreadLocalMap.Entry 的定义
static class Entry extends WeakReference<ThreadLocal<?>> {
    Object value;  // 强引用！
    Entry(ThreadLocal<?> k, Object v) {
        super(k);   // Key 是弱引用 —— GC 时可被回收
        value = v;   // Value 是强引用 —— GC 时不会被回收
    }
}

泄漏链路 (Leak Chain)：
 
Thread (线程池中长期存活)
  └── ThreadLocalMap
        └── Entry
              ├── Key (WeakRef) → null  (ThreadLocal 已被 GC)
              └── Value (StrongRef) → 大对象  ← 💀 内存泄漏！

// ✅ 正确的 ThreadLocal 使用模板
private static final ThreadLocal<UserContext> userCtx = new ThreadLocal<>();
 
public void handleRequest(Request req) {
    try {
        userCtx.set(buildContext(req));   // 设置值
        processBusinessLogic();           // 使用值
    } finally {
        userCtx.remove();                 // ⚡ 必须在 finally 中移除！
    }
}

public class InheritableDemo {
 
    // 可继承的 ThreadLocal
    private static final InheritableThreadLocal<String> traceId =
            new InheritableThreadLocal<>();
 
    public static void main(String[] args) {
        traceId.set("TRACE-001");                         // 父线程设置
 
        new Thread(() -> {
            // 子线程可以读到父线程的值（创建时从父线程拷贝过来的）
            System.out.println("子线程: " + traceId.get()); // TRACE-001
        }).start();
    }
}

import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
 
// final 类：禁止继承
public final class ImmutableOrder {
 
    private final String orderId;        // 不可变字段
    private final double amount;         // 不可变字段
    private final List<String> items;    // ⚠️ List 本身是可变的！
 
    public ImmutableOrder(String orderId, double amount, List<String> items) {
        this.orderId = orderId;
        this.amount = amount;
        // 🔑 防御性拷贝（Defensive Copy）：拷贝一份，而不是直接引用原 List
        // 否则外部仍然可以通过原 List 修改内部数据
        this.items = Collections.unmodifiableList(new ArrayList<>(items));
    }
 
    public String getOrderId() {
        return orderId;                  // String 本身不可变，直接返回
    }
 
    public double getAmount() {
        return amount;                   // 基本类型，直接返回
    }
 
    public List<String> getItems() {
        // 返回的是不可变视图，外部调用 add/remove 会抛异常
        return items;
    }
 
    // 如果需要"修改"，返回一个新对象（Copy-On-Write 风格）
    public ImmutableOrder withAmount(double newAmount) {
        return new ImmutableOrder(this.orderId, newAmount, this.items);
    }
}

// 一行代码定义不可变类
// 编译器自动生成：private final 字段、构造器、getter、equals、hashCode、toString
public record Point(int x, int y) { }
 
// 使用
Point p = new Point(3, 4);
System.out.println(p.x());      // 3（getter 方法名不带 get 前缀）
System.out.println(p.y());      // 4
// p.x = 10;                    // ❌ 编译错误，字段是 final 的

public record Order(String id, List<String> items) {
    // 紧凑构造器（Compact Constructor）—— 没有参数列表
    public Order {
        // 在赋值之前拦截，做防御性拷贝
        items = List.copyOf(items);   // List.copyOf 返回不可变 List
    }
}

public class ConfigHolder {
 
    // volatile 保证可见性，不可变对象保证状态一致性
    private volatile ImmutableConfig config;
 
    public ImmutableConfig getConfig() {
        return config;  // 任意线程安全读取，无需加锁
    }
 
    public void updateConfig(ImmutableConfig newConfig) {
        // 直接替换引用（引用赋值在 JVM 中是原子操作）
        this.config = newConfig;
    }
}
 
// 不可变配置
public record ImmutableConfig(String dbUrl, int poolSize, boolean enableCache) { }

// CopyOnWriteArrayList 的核心逻辑（简化版）
public class CopyOnWriteArrayList<E> {
 
    // volatile 数组引用
    private transient volatile Object[] array;
 
    // 读操作：直接读，无锁 ✅
    public E get(int index) {
        return (E) array[index];         // 读取的是当前数组快照
    }
 
    // 写操作：复制一份新数组，修改新数组，替换引用
    public boolean add(E e) {
        synchronized (lock) {            // 写加锁（写写互斥）
            Object[] old = array;
            int len = old.length;
            Object[] newArr = Arrays.copyOf(old, len + 1);  // 复制旧数组
            newArr[len] = e;                                 // 在新数组上修改
            array = newArr;                                  // 替换引用（原子操作）
            return true;
        }
    }
}

// ❌ 传统方式：平台线程在高并发下迅速耗尽资源
// 每个线程占用 ~1MB 栈内存，10000 个线程 = ~10GB 仅栈内存
ExecutorService executor = Executors.newFixedThreadPool(200); // 通常最多几百个
for (int i = 0; i < 10000; i++) {
    executor.submit(() -> {
        // 模拟一次数据库查询，线程阻塞等待 I/O
        Thread.sleep(Duration.ofMillis(100));
        return "result";
    });
}
// 200 个线程同时执行，剩余 9800 个任务在队列中等待
// 吞吐量上限 = 200 / 0.1s = 2000 req/s

// ✅ Virtual Thread 方式：轻松创建百万级并发
// 每个虚拟线程仅占 ~1-2 KB 内存（对比平台线程 ~1 MB）
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 1_000_000; i++) { // 一百万个并发任务！
        executor.submit(() -> {
            // 阻塞操作时，虚拟线程自动从载体线程(carrier)上卸载(unmount)
            // 载体线程被释放去执行其他虚拟线程
            Thread.sleep(Duration.ofMillis(100));
            return "result";
        });
    }
} // try-with-resources 自动等待所有任务完成

// Kotlin Coroutine 的 suspend 关键字显式标记可挂起函数
// 编译器在编译期就知道哪些函数可能暂停执行
suspend fun fetchUserFromDb(id: Long): User {
    // delay 是非阻塞的挂起函数，不占用底层线程
    delay(100) // 模拟 DB 查询
    return User(id, "Alice")
}
 
// 结构化并发：coroutineScope 确保所有子协程完成或取消
suspend fun loadDashboard(): Dashboard = coroutineScope {
    // async 并发启动多个协程，共享同一父 scope
    val userDeferred = async { fetchUserFromDb(1L) }       // 协程 1
    val ordersDeferred = async { fetchOrders(1L) }         // 协程 2
    val recommendDeferred = async { getRecommendations() } // 协程 3
 
    // 所有协程并发执行，await 等待结果
    Dashboard(
        user = userDeferred.await(),
        orders = ordersDeferred.await(),
        recommendations = recommendDeferred.await()
    )
    // 若任一协程异常，其余自动取消 —— 这就是结构化并发的力量
}

import kotlinx.coroutines.*
 
fun main() = runBlocking {
    // Dispatchers.Default — 用于 CPU 密集型任务
    // 线程数 = CPU 核心数，专为计算优化
    launch(Dispatchers.Default) {
        // 图像处理、数据加密、复杂算法等
        val result = heavyComputation()
    }
 
    // Dispatchers.IO — 用于 I/O 密集型任务
    // 线程数可弹性扩展（默认最多 64 或核心数，取较大值）
    launch(Dispatchers.IO) {
        // 数据库查询、文件读写、网络请求等
        val data = readFromDatabase()
    }
 
    // Dispatchers.Main — 用于 UI 更新（Android）
    // 仅在有 Main dispatcher 的环境中可用
    // launch(Dispatchers.Main) { updateUI(data) }
}

// === Java Virtual Thread ===
// 阻塞点是"隐式"的，你看不出 Thread.sleep 是否会让出载体线程
// 但 JVM 会在底层自动处理 mount/unmount
Thread.startVirtualThread(() -> {
    var user = userRepo.findById(1L);   // 阻塞调用 → JVM 自动 unmount
    var orders = orderRepo.findByUser(user); // 同上
    process(user, orders);
});

// === Kotlin Coroutine ===
// 挂起点是"显式"的，suspend 关键字清晰标记了可能暂停的函数
// 编译器在编译期就验证 suspend 函数只能在协程上下文中调用
suspend fun loadUserData() {
    val user = userRepo.findById(1L)         // suspend fun → 编译器已知
    val orders = orderRepo.findByUser(user)  // suspend fun → 编译器已知
    process(user, orders)
}

// Java 的 StructuredTaskScope（Preview API，Java 21+）
// 需要显式管理，语法相对冗长
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    // fork 创建子任务，运行在虚拟线程上
    Subtask<User> userTask = scope.fork(() -> findUser(id));
    Subtask<List<Order>> orderTask = scope.fork(() -> findOrders(id));
 
    scope.join();          // 等待所有子任务完成
    scope.throwIfFailed(); // 若有子任务失败，抛出异常
 
    // 所有子任务成功，获取结果
    return new Dashboard(userTask.get(), orderTask.get());
}
// ShutdownOnFailure: 任一子任务失败则取消其余 —— 类似 Kotlin 的行为

// Kotlin 的结构化并发 —— 语法天然集成，更简洁
suspend fun loadDashboard(id: Long): Dashboard = coroutineScope {
    val user = async { findUser(id) }      // 自动绑定到父 scope
    val orders = async { findOrders(id) }  // 自动绑定到父 scope
 
    Dashboard(user.await(), orders.await())
    // 若 findUser 抛异常，findOrders 自动被取消
    // 无需 try-with-resources，scope 自动管理生命周期
}

// ❌ 危险模式：synchronized + I/O = 线程固定
// 虚拟线程无法从 Carrier 上卸载，Carrier 被白白占用
public synchronized User findById(Long id) {
    // 此处发生阻塞 I/O，但因在 synchronized 内
    // 虚拟线程被 "pinned" 到 Carrier Thread，无法让出
    return jdbcTemplate.queryForObject(
        "SELECT * FROM users WHERE id = ?",
        userRowMapper, id
    );
}
 
// ✅ 修复方案（Java 21-23）：使用 ReentrantLock 替代 synchronized
// ReentrantLock 不使用 OS 原生 monitor，JVM 可正常 unmount 虚拟线程
private final ReentrantLock lock = new ReentrantLock();
 
public User findById(Long id) {
    lock.lock(); // 获取锁
    try {
        // 此处即便阻塞，虚拟线程也能正常 unmount
        return jdbcTemplate.queryForObject(
            "SELECT * FROM users WHERE id = ?",
            userRowMapper, id
        );
    } finally {
        lock.unlock(); // 确保释放锁，避免死锁
    }
}

// ❌ 踩踏效应：100 万个虚拟线程同时发起 DB 连接
// 数据库连接池（如 HikariCP 默认 10 连接）瞬间被击穿
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 1_000_000; i++) {
        executor.submit(() -> {
            // 100 万个虚拟线程同时尝试获取 DB 连接
            // 数据库: "ERROR: too many connections"
            return userRepo.findById(i);
        });
    }
}
 
// ✅ 修复方案：使用 Semaphore 作为"逻辑限流器"
// Semaphore 替代了平台线程时代 "线程池大小" 的天然限流作用
private static final Semaphore DB_LIMITER = new Semaphore(50); // 最多 50 个并发 DB 请求
 
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 1_000_000; i++) {
        final int userId = i;
        executor.submit(() -> {
            DB_LIMITER.acquire();  // 获取许可，超过 50 则等待
            try {
                return userRepo.findById(userId);
            } finally {
                DB_LIMITER.release(); // 释放许可
            }
        });
    }
}

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
 
public class VirtualThreadDemo {
 
    // 限流器：防止同时打开过多连接
    private static final Semaphore LIMITER = new Semaphore(100);
 
    public static void main(String[] args) throws Exception {
        // 创建 HttpClient，底层连接池由 JDK 管理
        var client = HttpClient.newHttpClient();
        var request = HttpRequest.newBuilder()
                .uri(URI.create("https://httpbin.org/delay/1")) // 模拟 1s 延迟的接口
                .timeout(Duration.ofSeconds(10))
                .build();
 
        var start = Instant.now(); // 记录开始时间
 
        // 使用 virtual-thread-per-task executor
        // 每个任务分配一个虚拟线程，无需预设池大小
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < 10_000; i++) {
                final int taskId = i;
                executor.submit(() -> {
                    LIMITER.acquire(); // 获取许可，控制并发度
                    try {
                        // 阻塞调用 → 虚拟线程自动让出 Carrier
                        var response = client.send(request,
                                HttpResponse.BodyHandlers.ofString());
                        if (taskId % 1000 == 0) { // 每 1000 个打印一次
                            System.out.printf("Task %d: status=%d%n",
                                    taskId, response.statusCode());
                        }
                    } finally {
                        LIMITER.release(); // 释放许可
                    }
                    return null;
                });
            }
        } // executor.close() 等待所有任务完成
 
        var elapsed = Duration.between(start, Instant.now());
        // 理论耗时：10000 / 100并发 * 1s ≈ 100s
        // 实际因网络抖动会有偏差
        System.out.printf("10000 requests completed in %d seconds%n",
                elapsed.toSeconds());
    }
}

# application.yml
spring:
  threads:
    virtual:
      enabled: true  # 就这一行！Tomcat 将使用虚拟线程处理请求

// Spring Boot 自动配置后，每个 HTTP 请求都运行在虚拟线程上
// 你的 Controller 代码无需任何修改
@RestController
public class UserController {
 
    @Autowired
    private UserRepository userRepo; // 传统 JDBC/JPA，阻塞式
 
    @GetMapping("/users/{id}")
    public User getUser(@PathVariable Long id) {
        // 这个阻塞调用现在运行在虚拟线程上
        // JVM 在 JDBC 等待数据库响应时自动 unmount 虚拟线程
        // Carrier Thread 被释放去处理其他请求
        return userRepo.findById(id)
                .orElseThrow(() -> new ResponseStatusException(
                        HttpStatus.NOT_FOUND));
    }
}
// 效果：不改一行业务代码，并发能力从数百提升到数万

// 一句话总结三者关系：
// Platform Thread  = 1:1 OS Thread（重量级，数千级）
// Virtual Thread   = M:N OS Thread（轻量级，百万级，JVM 调度，隐式挂起）
// Kotlin Coroutine = M:N OS Thread（轻量级，百万级，库层调度，显式挂起）

// ThreadPoolExecutor 的完整构造函数 —— 7 大核心参数
public ThreadPoolExecutor(
    int corePoolSize,           // 核心线程数：池中始终存活的线程数量（即使空闲也不回收）
    int maximumPoolSize,        // 最大线程数：池中允许存在的最大线程数量上限
    long keepAliveTime,         // 空闲存活时间：非核心线程空闲超过此时间后被回收
    TimeUnit unit,              // 时间单位：keepAliveTime 的单位（秒、毫秒等）
    BlockingQueue<Runnable> workQueue,  // 工作队列：核心线程满时，新任务在此排队等待
    ThreadFactory threadFactory,        // 线程工厂：自定义线程的创建方式（命名、优先级等）
    RejectedExecutionHandler handler    // 拒绝策略：队列满且线程数达上限时的兜底处理
) { ... }

// ❌ 危险示例 1：newFixedThreadPool
// 内部使用 LinkedBlockingQueue（无界队列），任务堆积会导致 OOM
ExecutorService fixed = Executors.newFixedThreadPool(10);
// 等价于：new ThreadPoolExecutor(10, 10, 0L, MILLISECONDS, new LinkedBlockingQueue<>())
 
// ❌ 危险示例 2：newCachedThreadPool
// maximumPoolSize 为 Integer.MAX_VALUE，可能无限创建线程导致 OOM
ExecutorService cached = Executors.newCachedThreadPool();
// 等价于：new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, SECONDS, new SynchronousQueue<>())
 
// ❌ 危险示例 3：newSingleThreadExecutor
// 同样是无界队列，且只有 1 个线程，任务严重堆积
ExecutorService single = Executors.newSingleThreadExecutor();
 
// ❌ 危险示例 4：newScheduledThreadPool
// maximumPoolSize 为 Integer.MAX_VALUE，延迟任务异常时可能无限创建线程
ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(5);

public class ThreadPoolSizing {
 
    public static void main(String[] args) {
        // 获取当前机器的 CPU 核心数
        int cpuCores = Runtime.getRuntime().availableProcessors();
        System.out.println("CPU 核心数: " + cpuCores); // 例如：8
 
        // ========== 场景 1：CPU 密集型（如加密计算）==========
        // 公式：N = cpuCores + 1
        int cpuIntensivePoolSize = cpuCores + 1;  // 8 + 1 = 9
        System.out.println("CPU 密集型线程数: " + cpuIntensivePoolSize);
 
        // ========== 场景 2：IO 密集型（如 HTTP 调用）==========
        // 假设：平均每个请求耗时 200ms，其中 CPU 计算 20ms，IO 等待 180ms
        double waitTime = 180.0;    // W：IO 等待时间（ms）
        double computeTime = 20.0;  // C：CPU 计算时间（ms）
        double targetUtilization = 0.8; // U_cpu：目标 CPU 利用率 80%
 
        // Goetz 公式：N = N_cpu × U_cpu × (1 + W/C)
        int ioIntensivePoolSize = (int) (cpuCores * targetUtilization * (1 + waitTime / computeTime));
        // 8 × 0.8 × (1 + 9) = 8 × 0.8 × 10 = 64
        System.out.println("IO 密集型线程数: " + ioIntensivePoolSize);
    }
}

// ========== JDK 内置的 4 种拒绝策略 ==========
 
// 1. AbortPolicy（默认）：直接抛出 RejectedExecutionException
//    适用于：关键业务，宁可报错也不能丢失任务
RejectedExecutionHandler abort = new ThreadPoolExecutor.AbortPolicy();
 
// 2. CallerRunsPolicy：由提交任务的调用者线程自己执行该任务
//    适用于：不允许丢弃任务的场景，同时可以对提交方施加背压
//    副作用：如果调用者是主线程，会阻塞主线程，降低整体吞吐
RejectedExecutionHandler callerRuns = new ThreadPoolExecutor.CallerRunsPolicy();
 
// 3. DiscardPolicy：静默丢弃被拒绝的任务，不抛异常
//    适用于：可以容忍少量数据丢失的非关键任务（如日志上报）
RejectedExecutionHandler discard = new ThreadPoolExecutor.DiscardPolicy();
 
// 4. DiscardOldestPolicy：丢弃队列中最老的任务，然后重新提交被拒绝的任务
//    适用于：最新数据更有价值的场景（如实时行情更新）
RejectedExecutionHandler discardOldest = new ThreadPoolExecutor.DiscardOldestPolicy();

// ========== 自定义拒绝策略：记录日志 + 降级处理 ==========
public class CustomRejectedHandler implements RejectedExecutionHandler {
 
    // 记录被拒绝的任务总数，用 AtomicLong 保证线程安全
    private final AtomicLong rejectedCount = new AtomicLong(0);
 
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 递增拒绝计数器
        long count = rejectedCount.incrementAndGet();
 
        // 记录告警日志，包含线程池当前的关键状态指标
        log.warn("[线程池过载] 第 {} 次拒绝任务 | 活跃线程: {} | 队列积压: {} | 已完成: {}",
                count,
                executor.getActiveCount(),       // 当前活跃线程数
                executor.getQueue().size(),       // 当前队列中等待的任务数
                executor.getCompletedTaskCount()  // 已完成任务总数
        );
 
        // 降级策略：将任务序列化后发送到消息队列，稍后重试
        messageQueue.send(new TaskWrapper(r));
 
        // 可选：触发告警（钉钉/企业微信/邮件通知）
        alertService.sendAlert("线程池拒绝任务累计: " + count);
    }
}

// ========== 自定义线程工厂：有意义的线程命名 ==========
public class NamedThreadFactory implements ThreadFactory {
 
    // 线程池编号计数器（全局），区分同类线程池的不同实例
    private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1);
 
    // 当前线程池内部的线程编号计数器
    private final AtomicInteger threadNumber = new AtomicInteger(1);
 
    // 线程名称前缀，如 "order-process" → "order-process-pool-1-thread-3"
    private final String namePrefix;
 
    // 是否为守护线程
    private final boolean daemon;
 
    public NamedThreadFactory(String bizName, boolean daemon) {
        this.daemon = daemon;
        // 组合格式：业务名-pool-编号-thread-
        this.namePrefix = bizName + "-pool-" + POOL_NUMBER.getAndIncrement() + "-thread-";
    }
 
    @Override
    public Thread newThread(Runnable r) {
        // 创建新线程，赋予格式化的名称
        Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement());
        // 设置为守护/用户线程
        t.setDaemon(daemon);
        // 设置统一优先级为 NORMAL（避免优先级反转）
        t.setPriority(Thread.NORM_PRIORITY);
        // 设置全局未捕获异常处理器，防止线程静默死亡
        t.setUncaughtExceptionHandler((thread, ex) -> {
            log.error("线程 [{}] 发生未捕获异常", thread.getName(), ex);
        });
        return t;
    }
}

// ========== 生产级线程池配置示例 ==========
public class ThreadPoolConfig {
 
    /**
     * 创建一个面向 IO 密集型任务的线程池
     * 场景：HTTP 外部调用、数据库查询等
     */
    public static ThreadPoolExecutor createIOIntensivePool(String bizName) {
        // 获取 CPU 核心数
        int cpuCores = Runtime.getRuntime().availableProcessors();
 
        // IO 密集型：核心线程数 = CPU 核心数 × 2
        int corePoolSize = cpuCores * 2;
 
        // 最大线程数 = 核心线程数 × 2，提供弹性应对突发流量
        int maxPoolSize = corePoolSize * 2;
 
        // 非核心线程空闲 60 秒后回收
        long keepAliveTime = 60L;
 
        // 有界队列，容量 200，防止无限堆积
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(200);
 
        // 自定义线程工厂，带业务名称
        ThreadFactory threadFactory = new NamedThreadFactory(bizName, false);
 
        // CallerRunsPolicy 拒绝策略：不丢弃任务，自动施加背压
        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
 
        // 构造线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,   // 核心线程数
                maxPoolSize,    // 最大线程数
                keepAliveTime,  // 空闲存活时间
                TimeUnit.SECONDS,
                workQueue,      // 工作队列
                threadFactory,  // 线程工厂
                handler         // 拒绝策略
        );
 
        // 预热核心线程：启动时就创建所有核心线程，避免首批请求冷启动延迟
        executor.prestartAllCoreThreads();
 
        // 允许核心线程超时回收（可选，适用于流量波动大的场景）
        // executor.allowCoreThreadTimeOut(true);
 
        return executor;
    }
 
    /**
     * 创建一个面向 CPU 密集型任务的线程池
     * 场景：数据加密、图像处理、复杂运算等
     */
    public static ThreadPoolExecutor createCPUIntensivePool(String bizName) {
        int cpuCores = Runtime.getRuntime().availableProcessors();
 
        // CPU 密集型：核心线程数 = CPU 核心数 + 1
        int corePoolSize = cpuCores + 1;
 
        // 最大线程数 = 核心线程数（CPU 密集型不需要额外扩展）
        int maxPoolSize = corePoolSize;
 
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(100);
 
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                0L,             // core == max，keepAlive 无意义
                TimeUnit.MILLISECONDS,
                workQueue,
                new NamedThreadFactory(bizName, false),
                new ThreadPoolExecutor.AbortPolicy() // CPU 密集型任务通常不可降级，直接报错
        );
 
        executor.prestartAllCoreThreads();
        return executor;
    }
}

// ========== 线程池运行时监控指标采集 ==========
public class ThreadPoolMonitor {
 
    private final ThreadPoolExecutor executor;         // 被监控的线程池
    private final ScheduledExecutorService scheduler;  // 定时调度器
 
    public ThreadPoolMonitor(ThreadPoolExecutor executor) {
        this.executor = executor;
        // 单线程的调度器，用于周期性采集指标
        this.scheduler = Executors.newSingleThreadScheduledExecutor(
                new NamedThreadFactory("pool-monitor", true) // 守护线程
        );
    }
 
    /**
     * 每隔 interval 秒采集一次线程池指标并输出
     */
    public void startMonitoring(long intervalSeconds) {
        scheduler.scheduleAtFixedRate(() -> {
            // -------- 线程相关指标 --------
            int poolSize = executor.getPoolSize();           // 当前线程池中的总线程数
            int activeCount = executor.getActiveCount();     // 当前正在执行任务的线程数
            int coreSize = executor.getCorePoolSize();       // 核心线程数（配置值）
            int maxSize = executor.getMaximumPoolSize();     // 最大线程数（配置值）
            int largestSize = executor.getLargestPoolSize(); // 历史峰值线程数
 
            // -------- 任务相关指标 --------
            long taskCount = executor.getTaskCount();            // 已提交任务总数
            long completedCount = executor.getCompletedTaskCount(); // 已完成任务总数
            int queueSize = executor.getQueue().size();          // 当前队列中等待的任务数
            int queueCapacity = executor.getQueue().remainingCapacity() + queueSize; // 队列总容量
 
            // -------- 计算关键比率 --------
            double threadUtilization = poolSize == 0 ? 0 :
                    (double) activeCount / poolSize * 100;       // 线程利用率 (%)
            double queueUtilization = queueCapacity == 0 ? 0 :
                    (double) queueSize / queueCapacity * 100;    // 队列使用率 (%)
 
            // -------- 日志输出 --------
            log.info("[线程池监控] 线程: {}/{}/{} (活跃/当前/最大) | " +
                            "利用率: {:.1f}% | 队列: {}/{} ({:.1f}%) | " +
                            "任务: {}/{} (完成/总计) | 历史峰值线程: {}",
                    activeCount, poolSize, maxSize,
                    threadUtilization,
                    queueSize, queueCapacity, queueUtilization,
                    completedCount, taskCount,
                    largestSize
            );
 
            // -------- 告警判断 --------
            if (queueUtilization > 80) {
                log.warn("[线程池告警] 队列使用率超过 80%，当前: {:.1f}%", queueUtilization);
            }
            if (threadUtilization > 90) {
                log.warn("[线程池告警] 线程利用率超过 90%，当前: {:.1f}%", threadUtilization);
            }
        }, 0, intervalSeconds, TimeUnit.SECONDS); // 立即开始，每 interval 秒执行一次
    }
}

// ========== 动态调参：运行时修改线程池参数 ==========
public class DynamicThreadPool {
 
    private final ThreadPoolExecutor executor;
 
    public DynamicThreadPool(ThreadPoolExecutor executor) {
        this.executor = executor;
    }
 
    /**
     * 动态调整核心线程数
     * ThreadPoolExecutor 内部会自动处理线程的增减：
     *   - 如果新值 > 旧值：立即创建新线程来处理队列中等待的任务
     *   - 如果新值 < 旧值：多余的核心线程在空闲后逐步被回收
     */
    public void setCorePoolSize(int newCoreSize) {
        log.info("调整核心线程数: {} → {}", executor.getCorePoolSize(), newCoreSize);
        executor.setCorePoolSize(newCoreSize);  // 实时生效，无需重启
    }
 
    /**
     * 动态调整最大线程数
     * 注意：新值不能小于当前的 corePoolSize
     */
    public void setMaxPoolSize(int newMaxSize) {
        log.info("调整最大线程数: {} → {}", executor.getMaximumPoolSize(), newMaxSize);
        executor.setMaximumPoolSize(newMaxSize);  // 实时生效
    }
 
    /**
     * 动态调整队列容量 —— 需要使用支持动态容量的队列
     * 注意：ArrayBlockingQueue 不支持动态修改容量！
     * 方案：使用 LinkedBlockingQueue 并通过反射修改 capacity 字段
     *       或者使用开源的 ResizableCapacityLinkedBlockingQueue
     */
    public void setQueueCapacity(int newCapacity) {
        BlockingQueue<Runnable> queue = executor.getQueue();
        if (queue instanceof ResizableCapacityLinkedBlockingQueue) {
            ((ResizableCapacityLinkedBlockingQueue<Runnable>) queue).setCapacity(newCapacity);
            log.info("调整队列容量 → {}", newCapacity);
        } else {
            log.warn("当前队列类型不支持动态调整容量: {}", queue.getClass().getSimpleName());
        }
    }
}

// ========== 配合 Nacos 配置中心实现动态线程池 ==========
@NacosConfigListener(dataId = "thread-pool-config.json", groupId = "DEFAULT_GROUP")
public void onConfigChanged(String newConfig) {
    // 解析新的配置
    ThreadPoolProperties props = JSON.parseObject(newConfig, ThreadPoolProperties.class);
 
    // 动态生效
    dynamicPool.setCorePoolSize(props.getCorePoolSize());
    dynamicPool.setMaxPoolSize(props.getMaxPoolSize());
    dynamicPool.setQueueCapacity(props.getQueueCapacity());
 
    log.info("线程池配置已动态更新: {}", props);
}

// ========== 线程池隔离配置 ==========
@Configuration
public class ThreadPoolIsolationConfig {
 
    /**
     * 快速响应池：处理用户 API 请求
     * 特点：线程多、队列短、快速失败
     */
    @Bean("apiThreadPool")
    public ThreadPoolExecutor apiThreadPool() {
        return new ThreadPoolExecutor(
                16, 32, 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100),                // 队列较短，避免排队
                new NamedThreadFactory("api-handler", false),
                new ThreadPoolExecutor.AbortPolicy()          // API 超载直接拒绝，返回 503
        );
    }
 
    /**
     * 批量处理池：处理报表导出、数据同步等
     * 特点：线程少、队列长、允许排队
     */
    @Bean("batchThreadPool")
    public ThreadPoolExecutor batchThreadPool() {
        return new ThreadPoolExecutor(
                4, 8, 120, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1000),               // 队列较长，允许批量排队
                new NamedThreadFactory("batch-worker", false),
                new ThreadPoolExecutor.CallerRunsPolicy()     // 批量任务不能丢
        );
    }
 
    /**
     * 通知发送池：处理短信、邮件、推送等异步通知
     * 特点：IO 密集、线程较多
     */
    @Bean("notifyThreadPool")
    public ThreadPoolExecutor notifyThreadPool() {
        return new ThreadPoolExecutor(
                8, 16, 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(500),
                new NamedThreadFactory("notify-sender", false),
                new ThreadPoolExecutor.DiscardOldestPolicy()  // 通知可丢弃旧的
        );
    }
}

// ========== 线程池优雅关闭 ==========
public class GracefulShutdown {
 
    /**
     * 两阶段优雅关闭线程池
     *
     * @param executor       要关闭的线程池
     * @param timeoutSeconds 等待超时时间（秒）
     */
    public static void shutdownGracefully(ThreadPoolExecutor executor, long timeoutSeconds) {
 
        // ===== 第一阶段：温柔关闭 =====
        // shutdown() 不接受新任务，但会把队列中的已有任务执行完毕
        executor.shutdown();
        log.info("线程池已调用 shutdown()，不再接受新任务");
 
        try {
            // 等待指定时间，让正在执行的任务和队列中的任务有机会完成
            boolean terminated = executor.awaitTermination(timeoutSeconds, TimeUnit.SECONDS);
 
            if (!terminated) {
                // ===== 第二阶段：强制关闭 =====
                // 如果超时后仍有任务未完成，强制中断所有线程
                log.warn("线程池在 {}s 内未完全关闭，开始强制终止...", timeoutSeconds);
 
                // shutdownNow() 返回队列中尚未执行的任务列表
                List<Runnable> unfinishedTasks = executor.shutdownNow();
                log.warn("被丢弃的队列任务数: {}", unfinishedTasks.size());
 
                // 再等一小段时间确认线程确实停止
                if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                    log.error("线程池无法完全关闭！可能存在不响应中断的任务");
                }
            }
        } catch (InterruptedException e) {
            // 当前线程在等待期间被中断，直接强制关闭
            log.warn("关闭线程被中断，执行强制关闭");
            executor.shutdownNow();
            // 恢复中断状态
            Thread.currentThread().interrupt();
        }
 
        log.info("线程池已关闭，isTerminated = {}", executor.isTerminated());
    }
}