并发不是开线程：从 JMM 到线程池与 CompletableFuture

线程读取变量：read → load → use
线程修改变量：assign → store → write

class Singleton {
    private static volatile Singleton instance; // 必须 volatile
    
    public static Singleton getInstance() {
        if (instance == null) {
            synchronized (Singleton.class) {
                if (instance == null) {
                    instance = new Singleton(); // 无 volatile 可能看到未初始化对象
                }
            }
        }
        return instance;
    }
}

class ImmutableObject {
    private final int x;
    
    public ImmutableObject(int x) {
        this.x = x; // final 字段的初始化保证可见性
    }
}

           [Java Thread]          [Java Thread]
               ↓   ↑                   ↓   ↑
           [工作内存]               [工作内存]
               ↓   ↑                   ↓   ↑
           [CPU 缓存]               [CPU 缓存]
               ↖   ↗               ↖   ↗
                 [主内存/RAM]

public class DeadLockDemo {
    private static Object resource1 = new Object();//资源 1
    private static Object resource2 = new Object();//资源 2

    public static void main(String[] args) {
        new Thread(() -> {
            synchronized (resource1) {
                System.out.println(Thread.currentThread() + "get resource1");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + "waiting get resource2");
                synchronized (resource2) {
                    System.out.println(Thread.currentThread() + "get resource2");
                }
            }
        }, "线程 1").start();

        new Thread(() -> {
            synchronized (resource2) {
                System.out.println(Thread.currentThread() + "get resource2");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + "waiting get resource1");
                synchronized (resource1) {
                    System.out.println(Thread.currentThread() + "get resource1");
                }
            }
        }, "线程 2").start();
    }
}

Thread[线程 1,5,main]get resource1
Thread[线程 2,5,main]get resource2
Thread[线程 1,5,main]waiting get resource2
Thread[线程 2,5,main]waiting get resource1

new Thread(() -> {
            synchronized (resource1) {
                System.out.println(Thread.currentThread() + "get resource1");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + "waiting get resource2");
                synchronized (resource2) {
                    System.out.println(Thread.currentThread() + "get resource2");
                }
            }
        }, "线程 2").start();

Thread[线程 1,5,main]get resource1
Thread[线程 1,5,main]waiting get resource2
Thread[线程 1,5,main]get resource2
Thread[线程 2,5,main]get resource1
Thread[线程 2,5,main]waiting get resource2
Thread[线程 2,5,main]get resource2

Process finished with exit code 0

public class VolatileAtomicityDemo {
    public volatile static int inc = 0;

    public void increase() {
        inc++;
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        VolatileAtomicityDemo volatileAtomicityDemo = new VolatileAtomicityDemo();
        for (int i = 0; i < 5; i++) {
            threadPool.execute(() -> {
                for (int j = 0; j < 500; j++) {
                    volatileAtomicityDemo.increase();
                }
            });
        }
        // 等待1.5秒，保证上面程序执行完成
        Thread.sleep(1500);
        System.out.println(inc);
        threadPool.shutdown();
    }
}

/**
  *  CAS
  * @param o         包含要修改field的对象
  * @param offset    对象中某field的偏移量
  * @param expected  期望值
  * @param update    更新值
  * @return          true | false
  */
public final native boolean compareAndSwapObject(Object o, long offset,  Object expected, Object update);

public final native boolean compareAndSwapInt(Object o, long offset, int expected,int update);

public final native boolean compareAndSwapLong(Object o, long offset, long expected, long update);

// 获取 Unsafe 实例
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long valueOffset;

static {
    try {
        // 获取“value”字段在AtomicInteger类中的内存偏移量
        valueOffset = unsafe.objectFieldOffset
            (AtomicInteger.class.getDeclaredField("value"));
    } catch (Exception ex) { throw new Error(ex); }
}
// 确保“value”字段的可见性
private volatile int value;

// 如果当前值等于预期值，则原子地将值设置为newValue
// 使用 Unsafe#compareAndSwapInt 方法进行CAS操作
public final boolean compareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

// **原子地**将当前值加 delta 并返回旧值
public final int getAndAdd(int delta) {
    return unsafe.getAndAddInt(this, valueOffset, delta);
}

// **原子地**将当前值加 1 并返回加之前的值（旧值）
// 使用 Unsafe#getAndAddInt 方法进行CAS操作。
public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}

// **原子地**将当前值减 1 并返回减之前的值（旧值）
public final int getAndDecrement() {
    return unsafe.getAndAddInt(this, valueOffset, -1);
}

// 原子地获取并增加整数值
public final int getAndAddInt(Object o, long offset, int delta) {
    int v;
    do {
        // 以 volatile 方式获取对象 o 在内存偏移量 offset 处的整数值
        v = getIntVolatile(o, offset);
    } while (!compareAndSwapInt(o, offset, v, v + delta));
    // 返回旧值
    return v;
}

synchronized void method() {
    //业务代码
}

synchronized static void method() {
    //业务代码
}

synchronized(this) {
    //业务代码
}

public class SynchronizedDemo {
    public void method() {
        synchronized (this) {
            System.out.println("synchronized 代码块");
        }
    }
}

public class SynchronizedDemo2 {
    public synchronized void method() {
        System.out.println("synchronized 方法");
    }
}

public class ReentrantLock implements Lock, java.io.Serializable {}

// 传入一个 boolean 值，true 时为公平锁，false 时为非公平锁
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

Atomic 原子类体系
├── 基本类型
│   ├── AtomicInteger
│   ├── AtomicLong
│   └── AtomicBoolean
│
├── 引用类型
│   ├── AtomicReference
│   ├── AtomicStampedReference（带版本号）
│   └── AtomicMarkableReference（带标记位）
│
├── 数组类型
│   ├── AtomicIntegerArray
│   ├── AtomicLongArray
│   └── AtomicReferenceArray
│
├── 字段更新器
│   ├── AtomicIntegerFieldUpdater
│   ├── AtomicLongFieldUpdater
│   └── AtomicReferenceFieldUpdater
│
└── 高性能计数器
    ├── LongAdder（Java8+）
    └── DoubleAdder（Java8+）

           +-----------------+
           | 读取内存值 V     |
           +-----------------+
                     |
                     v
+----------------------------------+
| 比较 V 与预期值 A                 |
| if (V == A) → 写入新值 B          | → 成功返回 true
| else        → 放弃操作            | → 失败返回 false
+----------------------------------+

AtomicInteger counter = new AtomicInteger(0);

// 原子递增
counter.incrementAndGet();  // 1

// 自定义更新
int result = counter.updateAndGet(x -> x * 2);  // 2

// 复合操作
boolean success = counter.compareAndSet(2, 5); // true

AtomicReference<User> userRef = new AtomicReference<>();
User oldUser = new User("Alice");
User newUser = new User("Bob");

// 原子替换
userRef.set(oldUser);
userRef.compareAndSet(oldUser, newUser); // 成功

LongAdder totalBytes = new LongAdder();

// 并发添加
parallelStream().forEach(i -> totalBytes.add(i));

// 获取最终结果（非实时）
long sum = totalBytes.sum();

AtomicStampedReference<Integer> money = new AtomicStampedReference<>(100, 0);

int[] stampHolder = new int[1];
int current = money.get(stampHolder); // 值=100, 版本=0

// 带版本号的CAS操作
money.compareAndSet(100, 200, stampHolder[0], stampHolder[0]+1);

✅ 计数器（访问量统计）
✅ 状态标志位（初始化标志）
✅ 对象属性原子更新
✅ 无锁数据结构（队列、栈）
✅ 实时性要求不高的统计

// 错误示例：复合操作仍需保护
AtomicInteger count = new AtomicInteger(0);
if (count.get() < 10) {  // 此处存在竞态条件！
    count.incrementAndGet();
}

// 正确做法：使用CAS循环
while (true) {
    int current = count.get();
    if (current >= 10) break;
    if (count.compareAndSet(current, current+1)) break;
}

public class LockFreeStack<T> {
    private AtomicReference<Node<T>> top = new AtomicReference<>();

    public void push(T item) {
        Node<T> newHead = new Node<>(item);
        Node<T> oldHead;
        do {
            oldHead = top.get();
            newHead.next = oldHead;
        } while (!top.compareAndSet(oldHead, newHead));
    }

    public T pop() {
        Node<T> oldHead;
        Node<T> newHead;
        do {
            oldHead = top.get();
            if (oldHead == null) return null;
            newHead = oldHead.next;
        } while (!top.compareAndSet(oldHead, newHead));
        return oldHead.item;
    }

    private static class Node<T> {
        final T item;
        Node<T> next;
        // 构造方法...
    }
}

class MetricCounter {
    private final AtomicLongArray counters;

    public MetricCounter(int size) {
        counters = new AtomicLongArray(size);
    }

    public void increment(int index) {
        counters.getAndIncrement(index);
    }

    public long getTotal() {
        long sum = 0;
        for (int i = 0; i < counters.length(); i++) {
            sum += counters.get(i);
        }
        return sum;
    }
}

AtomicInteger atomic = new AtomicInteger(5);

// 函数式更新
atomic.accumulateAndGet(3, Math::max); // 结果为5

// 延迟初始化
AtomicReference<ExpensiveObject> ref = new AtomicReference<>();
ExpensiveObject obj = ref.updateAndGet(
    o -> o != null ? o : createExpensiveObject()
);

LongAdder → 适合统计场景（写多读少）
LongAccumulator → 支持自定义累加规则

AtomicInteger atomicInt = new AtomicInteger(0);

// 基础操作
atomicInt.set(10);                          // 设置值
int val = atomicInt.get();                  // 获取值 → 10

// 原子运算
int newVal = atomicInt.incrementAndGet();    // 递增 → 11
int oldVal = atomicInt.getAndAdd(5);         // 返回旧值 → 11，新值 → 16

// CAS 操作
boolean success = atomicInt.compareAndSet(16, 20); // true

// 函数式更新
atomicInt.updateAndGet(x -> x * 2);          // 20 → 40

AtomicLong atomicLong = new AtomicLong(100L);

// 类似 AtomicInteger，支持大范围计数
atomicLong.addAndGet(50);                    // 150L
long current = atomicLong.getAndDecrement(); // 150 → 149

AtomicBoolean atomicBool = new AtomicBoolean(false);

// 原子切换状态
boolean old = atomicBool.getAndSet(true);    // false → true

// 条件更新
boolean updated = atomicBool.compareAndSet(true, false); // true

AtomicReference<String> ref = new AtomicReference<>("A");

// 更新引用
ref.set("B");
String oldVal = ref.getAndSet("C");          // "B" → "C"

// CAS 存在 ABA 问题
ref.compareAndSet("C", "D");                 // true

AtomicStampedReference<String> stampedRef = 
    new AtomicStampedReference<>("A", 0);

// 解决 ABA 问题
int[] stampHolder = new int[1];
String currentRef = stampedRef.get(stampHolder); // currentRef="A", stamp=0

// 更新时检查版本号
boolean success = stampedRef.compareAndSet(
    "A", "B", stampHolder[0], stampHolder[0] + 1); // 成功，版本号变为1

AtomicMarkableReference<String> markableRef = 
    new AtomicMarkableReference<>("A", false);

// 使用布尔标记
boolean[] markHolder = new boolean[1];
String ref = markableRef.get(markHolder);    // ref="A", mark=false

// 更新标记位
markableRef.attemptMark("A", true);          // 标记为 true

int[] arr = {1, 2, 3};
AtomicIntegerArray atomicArray = new AtomicIntegerArray(arr);

// 原子更新元素
atomicArray.set(0, 10);                      // 索引0 → 10
int val = atomicArray.getAndAdd(1, 5);       // 索引1: 2 → 7，返回旧值2

// CAS 操作
boolean updated = atomicArray.compareAndSet(2, 3, 30); // 索引2: 3 → 30

AtomicReferenceArray<String> refArray = 
    new AtomicReferenceArray<>(new String[5]);

refArray.set(0, "Java");
String old = refArray.getAndUpdate(0, s -> s + "8"); // "Java" → "Java8"

class Counter {
    volatile int count; // 必须是 volatile
}

Counter obj = new Counter();
AtomicIntegerFieldUpdater<Counter> updater = 
    AtomicIntegerFieldUpdater.newUpdater(Counter.class, "count");

updater.addAndGet(obj, 5); // obj.count = 5

class Node {
    volatile Node next; // 必须是 volatile
}

Node head = new Node();
AtomicReferenceFieldUpdater<Node, Node> updater = 
    AtomicReferenceFieldUpdater.newUpdater(Node.class, Node.class, "next");

updater.compareAndSet(head, null, new Node()); // 更新 head.next

LongAdder adder = new LongAdder();

// 并发累加（适合写多读少）
adder.add(10);
adder.increment();

// 获取总和（非精确实时值）
long sum = adder.sum();                      // 11

// 重置（不影响正在进行的累加）
adder.reset();                               // sum → 0

LongAccumulator accumulator = 
    new LongAccumulator((x, y) -> x * y, 1); // 初始值1，累乘器

accumulator.accumulate(3);                   // 1*3=3
accumulator.accumulate(5);                   // 3*5=15
long result = accumulator.get();             // 15

AtomicInteger ai = new AtomicInteger(0);

// 设置值（非原子）
ai.set(10); 

// 获取当前值
int current = ai.get(); // 10

// 原子设置并返回旧值
int old = ai.getAndSet(20); // old=10, ai=20

// 比较并交换（CAS）
boolean success = ai.compareAndSet(20, 30); // true

// 递增并获取新值
int newVal = ai.incrementAndGet(); // 31

// 获取并递增
int before = ai.getAndIncrement(); // 31 → ai=32

// 原子加法
int result = ai.addAndGet(5); // 32+5=37

// 自定义运算
int custom = ai.updateAndGet(x -> x * 2); // 37*2=74

// 累积计算（线程安全版本）
int accumulated = ai.accumulateAndGet(10, (prev, x) -> prev + x); // 74+10=84

// 延迟初始化（常用于单例）
AtomicInteger lazy = new AtomicInteger();
int initialized = lazy.updateAndGet(x -> x == 0 ? 100 : x);

LongAdder adder = new LongAdder();

// 递增（无返回值）
adder.increment();       // +1
adder.add(5);            // +5

// 递减
adder.decrement();       // -1

// 重置计数器
adder.reset();           // 归零

// 获取当前值（非原子快照）
long sum = adder.sum();  // 5

// 先加后获取（类似 getAndAdd）
long beforeAdd = adder.sumThenReset(); // 返回当前值并重置

// 与其他Adder合并
LongAdder another = new LongAdder();
another.add(3);
adder.add(another.sum()); // adder=8

// 高并发统计示例
LongAdder totalRequests = new LongAdder();
LongAdder failedRequests = new LongAdder();

// 请求处理线程
void handleRequest() {
    totalRequests.increment();
    try {
        processRequest();
    } catch (Exception e) {
        failedRequests.increment();
    }
}

// 输出统计
System.out.printf("成功率: %.2f%%%n", 
    100 * (totalRequests.sum() - failedRequests.sum()) / (double)totalRequests.sum());

@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
public class CounterBenchmark {
    private AtomicInteger atomic = new AtomicInteger();
    private LongAdder adder = new LongAdder();

    // 线程数模拟不同并发级别
    @Param({"1", "4", "8"})
    private int threads;

    @Benchmark
    public void atomicIncrement(Blackhole bh) {
        atomic.incrementAndGet();
    }

    @Benchmark
    public void adderIncrement(Blackhole bh) {
        adder.increment();
    }
}

需要原子变量吗？
├─ 是 → 写操作是否高频？
│   ├─ 是 → LongAdder
│   └─ 否 → AtomicInteger
└─ 否 → 考虑其他同步机制

// 1. 简单计数器
AtomicInteger counter = new AtomicInteger();
requestHandlers.forEach(h -> 
    h.setOnSuccess(() -> counter.incrementAndGet())
);

// 2. 状态标志位
AtomicInteger status = new AtomicInteger(0);
if (status.compareAndSet(0, 1)) {
    // 执行初始化操作
}

// 1. 实时数据统计
LongAdder bytesSent = new LongAdder();
networkService.addListener(bytes -> 
    bytesSent.add(bytes)
);

// 2. 高频计数器
LongAdder clickCounter = new LongAdder();
button.addClickListener(e -> 
    clickCounter.increment()
);

// 继承 Striped64 实现自定义分散算法
class CustomAdder extends Striped64 {
    public void increment() {
        Cell[] cs; long b, v; int m;
        if ((cs = cells) != null || 
            !casBase(b = base, b + 1)) {
            // 自定义冲突处理逻辑
        }
    }
}

ConcurrentHashMap<String, LongAdder> counterMap = new ConcurrentHashMap<>();

// 统计单词频率
words.forEach(word -> 
    counterMap.computeIfAbsent(word, k -> new LongAdder()).increment()
);

1. 定位过期槽位（staleSlot）
2. 向前扫描：找到最前端的连续过期 Entry 的起始位置（slotToExpunge）
3. 向后扫描：找到第一个非空 Entry 或数组末尾
   - 若找到匹配的 key，直接替换 value
   - 若找到新的过期 Entry，更新 slotToExpunge
4. 清理起始位置到当前槽位的过期 Entry
5. 调用 expungeStaleEntry 进行深度清理

private void replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot) {
    Entry[] tab = table;
    int len = tab.length;
    Entry e;

    // 向前扫描找到第一个过期 Entry 的位置
    int slotToExpunge = staleSlot;
    for (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len)) {
        if (e.get() == null) slotToExpunge = i;
    }

    // 向后扫描查找可替换的 Entry
    for (int i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();

        // 找到匹配的 key，替换 value 并交换槽位
        if (k == key) {
            e.value = value;
            tab[i] = tab[staleSlot];
            tab[staleSlot] = e;

            // 若向前扫描未找到其他过期 Entry，更新清理起点
            if (slotToExpunge == staleSlot) slotToExpunge = i;
            cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
            return;
        }

        // 发现新的过期 Entry，更新清理起点
        if (k == null && slotToExpunge == staleSlot) {
            slotToExpunge = i;
        }
    }

    // 未找到匹配 key，直接替换当前槽位
    tab[staleSlot].value = null;
    tab[staleSlot] = new Entry(key, value);

    // 触发深度清理
    if (slotToExpunge != staleSlot) {
        cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
    }
}

1. 清理指定槽位的过期 Entry
2. 向后扫描直到遇到空槽位：
   - 清理所有遇到的过期 Entry
   - 对有效 Entry 重新哈希定位
3. 返回下一个空槽位的位置

private int expungeStaleEntry(int staleSlot) {
    Entry[] tab = table;
    int len = tab.length;

    // 清理当前槽位
    tab[staleSlot].value = null;
    tab[staleSlot] = null;
    size--;

    Entry e;
    int i;
    // 向后扫描清理并重新哈希
    for (i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();
        if (k == null) { // 过期 Entry
            e.value = null;
            tab[i] = null;
            size--;
        } else { // 有效 Entry
            int h = k.threadLocalHashCode & (len - 1);
            if (h != i) { // 需要重新哈希
                tab[i] = null;
                while (tab[h] != null) h = nextIndex(h, len);
                tab[h] = e;
            }
        }
    }
    return i; // 返回下一个空槽位
}

初始容量 = 16
扩容阈值 = 初始容量 * 2/3 = 10
触发条件：size >= threshold

1. 全量清理过期 Entry（调用 expungeStaleEntries）
2. 若清理后 size >= threshold * 3/4，执行扩容
3. 容量翻倍（newLen = oldLen * 2）
4. 重新哈希所有有效 Entry 到新数组

private void resize() {
    Entry[] oldTab = table;
    int oldLen = oldTab.length;
    int newLen = oldLen * 2; // 容量翻倍
    Entry[] newTab = new Entry[newLen];
    int count = 0;

    // 重新哈希有效 Entry
    for (int j = 0; j < oldLen; j++) {
        Entry e = oldTab[j];
        if (e != null) {
            ThreadLocal<?> k = e.get();
            if (k == null) { // 跳过过期 Entry
                e.value = null; 
            } else {
                // 计算新位置
                int h = k.threadLocalHashCode & (newLen - 1);
                while (newTab[h] != null) h = nextIndex(h, newLen);
                newTab[h] = e;
                count++;
            }
        }
    }

    // 更新阈值
    setThreshold(newLen);
    size = count;
    table = newTab;
}

         [set()/get() 触发]
                 |
                 v
        发现过期 Entry（key=null）
                 |
         +-------+-------+
         |               |
         v               v
 replaceStaleEntry   expungeStaleEntry
 (替换并局部清理)      (深度清理)
         |               |
         +-------+-------+
                 |
                 v
           cleanSomeSlots
        (启发式清理后续槽位)
                 |
                 v
            [清理完成]

public class InheritableThreadLocalExample {
    private static InheritableThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>();

    public static void main(String[] args) {
        // 在主线程中设置 InheritableThreadLocal 的值
        inheritableThreadLocal.set("Hello from parent thread");

        // 创建子线程
        Thread childThread = new Thread(() -> {
            // 在子线程中获取 InheritableThreadLocal 的值
            String value = inheritableThreadLocal.get();
            System.out.println("Child thread value: " + value);
        });

        // 启动子线程
        childThread.start();

        // 等待子线程执行完毕
        try {
            childThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

import com.alibaba.ttl.TransmittableThreadLocal;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TransmittableThreadLocalExample {
    private static TransmittableThreadLocal<String> transmittableThreadLocal = new TransmittableThreadLocal<>();

    public static void main(String[] args) {
        // 在主线程中设置 TransmittableThreadLocal 的值
        transmittableThreadLocal.set("Hello from parent thread");

        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(2);

        // 提交任务到线程池
        executorService.submit(() -> {
            // 在子线程中获取 TransmittableThreadLocal 的值
            String value = transmittableThreadLocal.get();
            System.out.println("Child thread value: " + value);
        });

        // 关闭线程池
        executorService.shutdown();
    }
}

public ThreadPoolExecutor(int corePoolSize,//线程池的核心线程数量
                              int maximumPoolSize,//线程池的最大线程数
                              long keepAliveTime,//当线程数大于核心线程数时，多余的空闲线程存活的最长时间
                              TimeUnit unit,//时间单位
                              BlockingQueue<Runnable> workQueue,//任务队列，用来储存等待执行任务的队列
                              ThreadFactory threadFactory,//线程工厂，用来创建线程，一般默认即可
                              RejectedExecutionHandler handler//拒绝策略，当提交的任务过多而不能及时处理时，我们可以定制策略来处理任务

ThreadFactory threadFactory = new ThreadFactoryBuilder()
                        .setNameFormat(threadNamePrefix + "-%d")
                        .setDaemon(true).build();
ExecutorService threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程工厂，它设置线程名称，有利于我们定位问题。
 */
public final class NamingThreadFactory implements ThreadFactory {

    private final AtomicInteger threadNum = new AtomicInteger();
    private final String name;

    /**
     * 创建一个带名字的线程池生产工厂
     */
    public NamingThreadFactory(String name) {
        this.name = name;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r);
        t.setName(name + " [#" + threadNum.incrementAndGet() + "]");
        return t;
    }
}

import java.util.concurrent.*;

public class DynamicThreadPoolExample {
    public static void main(String[] args) throws InterruptedException {
        // 初始参数
        int corePoolSize = 2;
        int maximumPoolSize = 4;
        long keepAliveTime = 10;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(2);
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();

        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                workQueue,
                threadFactory,
                handler
        );

        // 提交初始任务
        for (int i = 1; i <= 6; i++) {
            final int taskId = i;
            try {
                executor.submit(() -> {
                    System.out.println("Task " + taskId + " is running on thread " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(2000); // 模拟任务执行时间
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            } catch (RejectedExecutionException e) {
                System.out.println("Task " + taskId + " was rejected");
            }
        }

        // 动态调整线程池参数
        Thread.sleep(3000); // 等待一段时间
        System.out.println("Adjusting thread pool parameters...");

        // 增加核心线程数和最大线程数
        executor.setCorePoolSize(4);
        executor.setMaximumPoolSize(6);

        // 提交更多任务
        for (int i = 7; i <= 12; i++) {
            final int taskId = i;
            try {
                executor.submit(() -> {
                    System.out.println("Task " + taskId + " is running on thread " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(2000); // 模拟任务执行时间
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            } catch (RejectedExecutionException e) {
                System.out.println("Task " + taskId + " was rejected");
            }
        }

        // 关闭线程池
        executor.shutdown();
    }
}

class ResizableCapacityLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> {
    private final AtomicInteger capacity; // 使用 AtomicInteger 来支持动态调整容量

    public ResizableCapacityLinkedBlockingQueue(int capacity) {
        super(capacity); // 初始化队列容量
        this.capacity = new AtomicInteger(capacity);
    }

    public synchronized void setCapacity(int newCapacity) {
        if (newCapacity < size()) {
            throw new IllegalArgumentException("New capacity cannot be less than current size");
        }
        this.capacity.set(newCapacity);
    }

    public int getCapacity() {
        return capacity.get();
    }

    @Override
    public boolean offer(E e) {
        // 检查当前队列大小是否超过容量
        if (size() >= capacity.get()) {
            return false; // 队列已满，拒绝添加
        }
        return super.offer(e);
    }

    @Override
    public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
        // 检查当前队列大小是否超过容量
        if (size() >= capacity.get()) {
            return false; // 队列已满，拒绝添加
        }
        return super.offer(e, timeout, unit);
    }

    @Override
    public void put(E e) throws InterruptedException {
        // 检查当前队列大小是否超过容量
        while (size() >= capacity.get()) {
            Thread.yield(); // 等待队列有空闲空间
        }
        super.put(e);
    }
}