static final class Node {
    // 标记：共享模式 / 独占模式
    static final Node SHARED = new Node();
    static final Node EXCLUSIVE = null;

    // waitStatus 的四种取值（0 是第三种：新建或已完成）
    static final int CANCELLED  =  1;  // 等待超时或被中断，需要移出队列
    static final int SIGNAL     = -1;  // 后继节点在等我释放/取消，我必须唤醒它
    static final int CONDITION  = -2;  // 节点正躺在 Condition 的等待队列里
    static final int PROPAGATE  = -3;  // 共享模式下，唤醒需要向后持续传播

    volatile int waitStatus;
    volatile Node prev;
    volatile Node next;
    volatile Thread thread;      // 节点代表的线程
    Node nextWaiter;             // Condition 队列的单向链，或 SHARED 标记
}

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&                                  // ① 先抢一次
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))       // ② 抢不到：入队 → 排队等待
        selfInterrupt();                                     // ③ 等待期间被中断过，补上中断标记
}

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 只有前驱是 head 才尝试获取：保证 FIFO 公平，避免队尾插队
            if (p == head && tryAcquire(arg)) {
                setHead(node);            // 自己成为 head，thread 置 null
                p.next = null;            // 帮助 GC 回收旧 head
                failed = false;
                return interrupted;
            }
            // 判断是否需要 park；返回 true 就 park，醒来后检查中断位
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;       // 注意：不抛异常，只记账，最后补中断
        }
    } finally {
        if (failed) cancelAcquire(node);  // 异常才取消排队
    }
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;                      // 前驱承诺会唤醒我，可以放心睡
    if (ws > 0) {
        // 前驱已取消，向前跳过所有 CANCELLED 节点，再修正链表
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 前驱是 0 或 PROPAGATE：用 CAS 把它改成 SIGNAL，下次循环再 park
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;

/**
 * 基于 AQS 的不可重入独占锁（Mutex）
 * 约定：state == 0 表示未锁定，state == 1 表示已锁定
 */
public class MutexLock implements java.io.Serializable {
    private static final long serialVersionUID = 1L;

    /** 静态内部类：只重写 tryAcquire / tryRelease / isHeldExclusively */
    private static class Sync extends AbstractQueuedSynchronizer {
        // 是否被占用
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        // state 为 0 时用 CAS 抢成 1，抢到即获得锁；不可重入：已占用直接失败
        @Override
        protected boolean tryAcquire(int acquires) {
            assert acquires == 1;
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread()); // 记录持有线程
                return true;
            }
            return false;
        }

        // 释放：state 置 0，必须先把持有者清空再 setState，保证可见顺序
        @Override
        protected boolean tryRelease(int releases) {
            assert releases == 1;
            if (getState() == 0) throw new IllegalMonitorStateException();
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        Condition newCondition() { return new ConditionObject(); }
    }

    private final Sync sync = new Sync();

    public void lock()              { sync.acquire(1); }
    public void unlock()            { sync.release(1); }
    public boolean tryLock()        { return sync.tryAcquire(1); }
    public boolean isLocked()       { return sync.isHeldExclusively(); }

    /** 可中断地获取锁，等待期间被 interrupt 会抛异常 */
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    /** 带超时的尝试，arg 为 1，超时时间由 AQS 的 nanosTimeout 处理 */
    public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }

    public Condition newCondition() { return sync.newCondition(); }

    // ---- 测试：10 个线程各加 1000 次，结果应为 10000 ----
    public static void main(String[] args) throws InterruptedException {
        MutexLock lock = new MutexLock();
        int[] counter = {0};
        Thread[] ts = new Thread[10];
        for (int i = 0; i < ts.length; i++) {
            ts[i] = new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    lock.lock();
                    try {
                        counter[0]++;          // 临界区：非原子操作
                    } finally {
                        lock.unlock();         // 必须放 finally，防止异常导致锁泄露
                    }
                }
            });
            ts[i].start();
        }
        for (Thread t : ts) t.join();
        System.out.println("counter = " + counter[0]);   // 稳定输出 10000
    }
}

public interface Lock {
    void lock();                                        // 拿不到就阻塞，不可中断
    void lockInterruptibly() throws InterruptedException; // 阻塞但响应中断
    boolean tryLock();                                  // 立即返回成败，不阻塞
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException; // 限时尝试
    void unlock();                                      // 必须释放，否则死锁
    Condition newCondition();                           // 绑定一个等待队列
}

// ReentrantLock.NonfairSync.tryAcquire（JDK 8 精简版）
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 无锁：CAS 抢一下，抢到就把持有者设为自己
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    } else if (current == getExclusiveOwnerThread()) {
        // 已锁且是自己：重入，state 累加（int 溢出会抛 Error，故上限极大）
        int nextc = c + acquires;
        if (nextc < 0) throw new Error("Maximum lock count exceeded");
        setState(nextc);      // 只有自己能改，无需 CAS
        return true;
    }
    return false;             // 锁被别人持有
}

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();   // 未持有却释放，直接报错
    boolean free = false;
    if (c == 0) {            // 完全释放：次数归零才算真正解锁
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

// 非公平：上来就抢，不管队里有没有人
static final class NonfairSync extends Sync {
    protected boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);     // 见 3.2，直接 CAS
    }
}

// 公平：先看看前面有没有排队的，有就老实排
static final class FairSync extends Sync {
    protected boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // hasQueuedPredecessors：队列里是否存在着比我更早的等待者
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        } else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0) throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}

// 判断"我前面是否有人在排队"：队空 || 队首就是我自己 || 队首线程是当前线程
public final boolean hasQueuedPredecessors() {
    Node t = tail, h = head, s;
    return h != t &&                                  // 队列非空
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

final boolean transferAfterCancelledWait(Node node) {
    // 情况一：正常被 signal（状态已被改成 0），CAS 回 CONDITION 失败，说明有人动了它
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        enq(node);          // 超时或被中断：自己把自己塞回同步队列
        return true;        // true 表示"中断发生在 signal 之前"
    }
    // 情况二：已经被 signal 了（状态已是 0），等它迁移完成即可
    while (!isOnSyncQueue(node))
        Thread.yield();     // signal 尚未入队，让步自旋等一会
    return false;           // false 表示"中断发生在 signal 之后"
}

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 有界缓冲区：经典生产者-消费者
 * 用两个 Condition 分别挂起"生产者"和"消费者"，实现精准唤醒
 */
public class BoundedBuffer<T> {
    private final Object[] items;
    private int putPtr, takePtr, count;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull  = lock.newCondition();  // 队列不满
    private final Condition notEmpty = lock.newCondition();  // 队列不空

    @SuppressWarnings("unchecked")
    public BoundedBuffer(int capacity) { items = new Object[capacity]; }

    public void put(T x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length)   // 必须用 while，防止虚假唤醒
                notFull.await();            // 释放锁并挂起，被唤醒后重新抢锁
            items[putPtr] = x;
            if (++putPtr == items.length) putPtr = 0;
            count++;
            notEmpty.signal();              // 只唤醒一个消费者，不用 signalAll
        } finally {
            lock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0)
                notEmpty.await();
            T x = (T) items[takePtr];
            if (++takePtr == items.length) takePtr = 0;
            count--;
            notFull.signal();               // 只唤醒一个生产者
            return x;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        BoundedBuffer<Integer> buf = new BoundedBuffer<>(5);
        // 生产者：每 300ms 放一个数
        new Thread(() -> {
            for (int i = 1; i <= 10; i++) {
                try { buf.put(i); System.out.println("put " + i); Thread.sleep(300); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }
        }, "producer").start();
        // 消费者：每 800ms 取一个，必然触发 notFull.await()
        new Thread(() -> {
            for (int i = 1; i <= 10; i++) {
                try { System.out.println("take " + buf.take()); Thread.sleep(800); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }
        }, "consumer").start();
    }
}

import java.util.concurrent.locks.ReentrantReadWriteLock;

/** 缓存示例：演示锁降级（写锁 → 读锁 → 释放写锁） */
public class CacheDemo {
    private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    private final ReentrantReadWriteLock.ReadLock  readLock  = rwl.readLock();
    private final ReentrantReadWriteLock.WriteLock writeLock = rwl.writeLock();
    private volatile boolean cacheValid;
    private Object data;

    public Object getData() {
        readLock.lock();
        if (!cacheValid) {
            // 必须先释放读锁，否则直接获取写锁会死锁（读锁不支持升级为写锁）
            readLock.unlock();
            writeLock.lock();
            try {
                if (!cacheValid) {          // 双重检查，防止重复加载
                    data = loadFromDb();
                    cacheValid = true;
                }
                // ↓↓↓ 锁降级开始：在持有写锁的情况下获取读锁 ↓↓↓
                readLock.lock();
            } finally {
                writeLock.unlock();         // 降级为读锁，其它读线程可进来
            }
        }
        try {
            return data;                    // 仍在读锁保护下返回，期间不会有写线程改它
        } finally {
            readLock.unlock();
        }
    }

    private Object loadFromDb() {
        System.out.println(Thread.currentThread().getName() + " 加载数据库...");
        return new Object();
    }
}

import java.util.concurrent.locks.StampedLock;

/** StampedLock 乐观读：读多写少场景下几乎无锁开销 */
public class Point {
    private double x, y;
    private final StampedLock sl = new StampedLock();

    // 写：独占，返回 stamp 用于释放
    public void move(double deltaX, double deltaY) {
        long stamp = sl.writeLock();
        try {
            x += deltaX;
            y += deltaY;
        } finally {
            sl.unlockWrite(stamp);
        }
    }

    // 乐观读：不加锁，靠 validate 兜底
    public double distanceFromOrigin() {
        long stamp = sl.tryOptimisticRead();   // 拿到"版本戳"，开销极小
        double currentX = x, currentY = y;     // 读字段（注意：这里要读 volatile 或用局部变量拷贝）
        if (!sl.validate(stamp)) {             // 期间被写过，数据可能不一致
            stamp = sl.readLock();             // 升级为悲观读锁
            try {
                currentX = x;
                currentY = y;
            } finally {
                sl.unlockRead(stamp);
            }
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }

    /** 悲观读 + 条件等待的写法（StampedLock 不支持 Condition！） */
    public double pessimisticDistance() {
        long stamp = sl.readLock();
        try {
            return Math.sqrt(x * x + y * y);
        } finally {
            sl.unlockRead(stamp);
        }
    }
}

import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.TimeUnit;

/** 复现 ReentrantLock 死锁：两个线程互相持有对方需要的锁 */
public class LockDeadlockDemo {
    private static final ReentrantLock lockA = new ReentrantLock();
    private static final ReentrantLock lockB = new ReentrantLock();

    public static void main(String[] args) throws InterruptedException {
        // 线程 1：先 A 后 B
        new Thread(() -> {
            lockA.lock();
            try {
                TimeUnit.SECONDS.sleep(1);       // 制造时间窗口，让对方也拿到 B
                System.out.println("T1 尝试获取 lockB...");
                lockB.lock();
                try { System.out.println("T1 拿到两把锁"); }
                finally { lockB.unlock(); }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally { lockA.unlock(); }
        }, "Thread-A-B").start();

        // 线程 2：先 B 后 A
        new Thread(() -> {
            lockB.lock();
            try {
                TimeUnit.SECONDS.sleep(1);
                System.out.println("T2 尝试获取 lockA...");
                lockA.lock();                    // 死在这里
                try { System.out.println("T2 拿到两把锁"); }
                finally { lockA.unlock(); }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally { lockB.unlock(); }
        }, "Thread-B-A").start();

        TimeUnit.SECONDS.sleep(3);
        System.out.println("主线程退出，请用 jstack <pid> 观察两个 WAITING (parking) 线程");
    }
}

// Unsafe 中的三个核心 CAS 方法（native，最终由 CPU 指令实现）
public final native boolean compareAndSwapObject(Object o, long offset, Object expected, Object x);
public final native boolean compareAndSwapInt (Object o, long offset, int  expected, int  x);
public final native boolean compareAndSwapLong(Object o, long offset, long expected, long x);

import java.util.concurrent.atomic.AtomicStampedReference;

/** ABA 问题与 AtomicStampedReference 的解决 */
public class ABADemo {
    public static void main(String[] args) throws InterruptedException {
        // 普通引用：无法察觉 A -> B -> A 的变化
        AtomicStampedReference<Integer> ref =
                new AtomicStampedReference<>(100, 1);   // 初始值 100，版本戳 1

        // 线程 t1：想把 100 改成 101，但它中间睡了 1 秒
        Thread t1 = new Thread(() -> {
            int stamp = ref.getStamp();               // 读到版本 1
            System.out.println("t1 读到 stamp=" + stamp);
            try { Thread.sleep(1000); } catch (InterruptedException e) { return; }
            // 期望：值仍是 100 且版本仍是 1 → 失败，因为版本已被 t2 改过
            boolean ok = ref.compareAndSet(100, 101, stamp, stamp + 1);
            System.out.println("t1 CAS 结果 = " + ok + "，当前值 = " + ref.getReference()
                    + "，当前版本 = " + ref.getStamp());
        }, "t1");

        // 线程 t2：制造 ABA —— 100 -> 101 -> 100，版本号从 1 变到 3
        Thread t2 = new Thread(() -> {
            ref.compareAndSet(100, 101, ref.getStamp(), ref.getStamp() + 1);
            System.out.println("t2 第一次修改，值=" + ref.getReference() + " 版本=" + ref.getStamp());
            ref.compareAndSet(101, 100, ref.getStamp(), ref.getStamp() + 1);
            System.out.println("t2 第二次修改（改回 100），值=" + ref.getReference()
                    + " 版本=" + ref.getStamp());
        }, "t2");

        t1.start();
        Thread.sleep(100);      // 确保 t1 先读到 stamp
        t2.start();
        t1.join(); t2.join();
        // 输出：t1 CAS 结果 = false —— 值虽然回到 100，但版本戳不匹配，操作被拒绝
    }
}

// JDK 8 AtomicInteger 源码（去掉注释）
public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}

// Unsafe.getAndAddInt：经典的 CAS 自旋循环
public final int getAndAddInt(Object o, long offset, int delta) {
    int v;
    do {
        v = this.getIntVolatile(o, offset);       // 1. volatile 读当前值
    } while (!this.compareAndSwapInt(o, offset, v, v + delta)); // 2. CAS 失败就重来
    return v;                                      // 3. 返回旧值
}

// Striped64 中的 Cell 定义：注意 @Contended
@sun.misc.Contended static final class Cell {
    volatile long value;
    Cell(long x) { value = x; }
    final boolean cas(long cmp, long val) {
        return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
    }
}

// LongAdder.add 的核心逻辑（伪代码化）
public void add(long x) {
    Cell[] cs; long b, v; int m; Cell c;
    if ((cs = cells) != null || !casBase(b = base, b + x)) {  // 先试 base
        boolean uncontended = true;
        if (cs == null || (m = cs.length - 1) < 0 ||
            (c = cs[getProbe() & m]) == null ||
            !(uncontended = c.cas(v = c.value, v + x)))
            longAccumulate(x, null, uncontended);   // 兜底：初始化/扩容/换槽
    }
}

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;

/** ConcurrentHashMap 常用原子复合操作与统计 */
public class ConcurrentHashMapDemo {
    public static void main(String[] args) {
        ConcurrentHashMap<String, LongAdder> wordCount = new ConcurrentHashMap<>();

        // 1. putIfAbsent：不存在才放，原子操作，避免"先 get 判断再 put"的竞态
        wordCount.putIfAbsent("java", new LongAdder());
        wordCount.get("java").increment();

        // 2. computeIfAbsent：更推荐的写法，一行搞定（注意不要在 lambda 里做耗时操作，
        //    因为 compute 会锁住当前桶）
        wordCount.computeIfAbsent("juc", k -> new LongAdder()).increment();
        wordCount.computeIfAbsent("juc", k -> new LongAdder()).increment();

        // 3. 遍历：弱一致迭代器，不会抛 ConcurrentModificationException
        wordCount.forEach((k, v) -> System.out.println(k + " -> " + v.sum()));

        // 4. 统计：mappingCount 返回 long，比 size() 的 int 更安全
        System.out.println("总 key 数 = " + wordCount.mappingCount());

        // 5. 批量：reduce / search / forEach 支持并行阈值
        System.out.println("总次数 = " + wordCount.reduceValuesToLong(1L, LongAdder::sum));

        // 6. 注意：value 不能为 null，会直接 NPE
        // wordCount.put("null-value-demo", null);  // java.lang.NullPointerException
    }
}

import java.util.concurrent.CopyOnWriteArrayList;

/** CopyOnWriteArrayList：读多写极少场景；演示迭代器的弱一致性 */
public class CopyOnWriteDemo {
    public static void main(String[] args) throws InterruptedException {
        CopyOnWriteArrayList<String> listeners = new CopyOnWriteArrayList<>();
        listeners.add("listener-1");
        listeners.add("listener-2");

        // 迭代过程中被修改：不会抛 ConcurrentModificationException
        Thread writer = new Thread(() -> {
            try { Thread.sleep(10); } catch (InterruptedException e) { return; }
            listeners.add("listener-3");   // 复制新数组，老迭代器看不到
            System.out.println("写线程已添加 listener-3");
        });
        writer.start();

        for (String s : listeners) {       // 快照迭代，输出 1、2（看不到 3）
            System.out.println("遍历到：" + s);
            Thread.sleep(50);
        }
        writer.join();
        System.out.println("遍历结束后 size = " + listeners.size());  // 3
    }
}

import java.util.concurrent.*;

/** 用 DelayQueue 实现订单 30 分钟未支付自动关闭 */
public class OrderTimeoutDemo {

    /** 延时任务：实现 Delayed 接口 */
    static class OrderTask implements Delayed {
        private final String orderId;
        private final long expireAt;   // 到期时间戳（毫秒）

        public OrderTask(String orderId, long delayMillis) {
            this.orderId = orderId;
            this.expireAt = System.currentTimeMillis() + delayMillis;
        }

        // 返回剩余延时，<=0 表示到期，可从队列取出
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(expireAt - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        // 按到期时间排序，DelayQueue 内部是优先队列，靠它决定出队顺序
        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.expireAt, ((OrderTask) o).expireAt);
        }

        public String getOrderId() { return orderId; }
    }

    public static void main(String[] args) throws InterruptedException {
        DelayQueue<OrderTask> delayQueue = new DelayQueue<>();

        // 模拟下单：三个订单，分别 2s、5s、3s 后超时
        delayQueue.put(new OrderTask("ORDER-001", 2_000));
        delayQueue.put(new OrderTask("ORDER-002", 5_000));
        delayQueue.put(new OrderTask("ORDER-003", 3_000));
        System.out.println("三个订单已下单，等待超时关闭...");

        // 守护线程消费到期订单
        Thread closer = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    OrderTask task = delayQueue.take();   // 只有到期才会返回
                    // 真实业务：先查订单状态，仍为"待支付"才关闭并释放库存
                    System.out.println("[" + System.currentTimeMillis() % 100_000
                            + "] 订单 " + task.getOrderId() + " 超时未支付，执行关闭");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }, "order-timeout-closer");
        closer.setDaemon(true);
        closer.start();

        closer.join(6_000);
    }
}

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

/** BlockingQueue 实现生产者-消费者的三种写法对比 */
public class ProducerConsumerDemo {

    /** 写法一：put/take —— 全自动阻塞，代码最少，最常用 */
    static void style1() {
        BlockingQueue<String> q = new ArrayBlockingQueue<>(10);
        new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    q.put("item-" + i);          // 满了自动等
                    System.out.println("生产 item-" + i);
                }
            } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        }).start();
        new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    System.out.println("消费 " + q.take());   // 空了自动等
                }
            } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        }).start();
    }

    /** 写法二：offer/poll 超时 —— 可做退避、可响应关闭信号，适合需要优雅退出的服务 */
    static void style2() {
        BlockingQueue<String> q = new LinkedBlockingQueue<>();
        volatile boolean[] running = {true};
        new Thread(() -> {
            while (running[0]) {
                try {
                    String item = q.poll(1, TimeUnit.SECONDS);  // 每秒醒来检查一次
                    if (item != null) System.out.println("消费 " + item);
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
            }
            System.out.println("消费者退出");
        }).start();
        try {
            q.offer("a", 2, TimeUnit.SECONDS);
            q.offer("b", 2, TimeUnit.SECONDS);
            Thread.sleep(1200);
            running[0] = false;                                  // 发关闭信号
        } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }

    /** 写法三：多生产多消费 + 毒丸（Poison Pill）优雅退出 */
    static void style3() throws InterruptedException {
        int producers = 2, consumers = 3;
        BlockingQueue<String> q = new LinkedBlockingQueue<>(20);
        String POISON = "POISON_PILL";
        AtomicInteger remainingConsumers = new AtomicInteger(consumers);
        CountDownLatch done = new CountDownLatch(consumers);

        for (int c = 0; c < consumers; c++) {
            new Thread(() -> {
                try {
                    while (true) {
                        String item = q.take();
                        if (POISON.equals(item)) {
                            // 收到毒丸：把自己那份"死亡信号"再放回队列，唤醒其它消费者
                            if (remainingConsumers.decrementAndGet() > 0) q.put(POISON);
                            break;
                        }
                        System.out.println(Thread.currentThread().getName() + " 消费 " + item);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    done.countDown();
                }
            }, "consumer-" + c).start();
        }

        for (int p = 0; p < producers; p++) {
            final int pid = p;
            new Thread(() -> {
                for (int i = 0; i < 6; i++) {
                    try { q.put("P" + pid + "-item" + i); } catch (InterruptedException e) { return; }
                }
            }, "producer-" + p).start();
        }

        Thread.sleep(500);
        q.put(POISON);              // 主线程投递一枚毒丸即可链式关闭所有消费者
        done.await();
        System.out.println("所有消费者已优雅退出");
    }

    public static void main(String[] args) throws InterruptedException {
        style1(); Thread.sleep(500);
        System.out.println("--- 写法二 ---"); style2(); Thread.sleep(1500);
        System.out.println("--- 写法三 ---"); style3();
    }
}

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

/**
 * 用 CountDownLatch 做接口并发压测
 * 两个门闩：startGate 让所有线程同时发令，endGate 等所有线程跑完再统计
 */
public class ConcurrentPressureTest {
    public static void main(String[] args) throws InterruptedException {
        int threadCount = 100;
        ExecutorService pool = Executors.newFixedThreadPool(32);
        CountDownLatch startGate = new CountDownLatch(1);              // 发令枪
        CountDownLatch endGate   = new CountDownLatch(threadCount);    // 终点线
        AtomicInteger success = new AtomicInteger();
        AtomicInteger failure = new AtomicInteger();
        AtomicLongHolder totalCost = new AtomicLongHolder();

        for (int i = 0; i < threadCount; i++) {
            pool.submit(() -> {
                try {
                    startGate.await();               // 全部就位，等发令
                    long start = System.currentTimeMillis();
                    boolean ok = mockCallApi();      // 被测接口
                    long cost = System.currentTimeMillis() - start;
                    totalCost.add(cost);
                    if (ok) success.incrementAndGet(); else failure.incrementAndGet();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    endGate.countDown();             // 每跑完一个减一
                }
            });
        }

        long begin = System.currentTimeMillis();
        startGate.countDown();                       // 发令！
        endGate.await();                             // 等全部结束
        long wall = System.currentTimeMillis() - begin;

        System.out.println("线程数 = " + threadCount);
        System.out.println("成功 = " + success.get() + "，失败 = " + failure.get());
        System.out.println("总耗时 = " + wall + " ms");
        System.out.println("平均 RT = " + (totalCost.sum() / threadCount) + " ms");
        System.out.println("估算 QPS = " + (threadCount * 1000L / wall));
        pool.shutdown();
    }

    /** 模拟接口调用：这里替换为真实 HTTP/RPC 调用即可 */
    private static boolean mockCallApi() throws InterruptedException {
        TimeUnit.MILLISECONDS.sleep(50 + (long) (Math.random() * 50));
        return Math.random() > 0.05;                 // 95% 成功率
    }

    /** 简易 long 累加器（演示用，生产环境直接用 LongAdder） */
    static class AtomicLongHolder {
        private final LongAdder adder = new LongAdder();
        void add(long v) { adder.add(v); }
        long sum() { return adder.sum(); }
    }
}

import java.util.concurrent.*;

/** CyclicBarrier：多线程分阶段计算，每阶段汇总一次 */
public class CyclicBarrierDemo {
    public static void main(String[] args) {
        int parties = 3;
        int[][] matrix = {
            {1, 2, 3},
            {4, 5, 6},
            {7, 8, 9}
        };
        int[] partial = new int[parties];

        // barrierAction：最后到达的线程执行汇总
        CyclicBarrier barrier = new CyclicBarrier(parties, () -> {
            int sum = 0;
            for (int p : partial) sum += p;
            System.out.println("本阶段汇总结果 = " + sum);
        });

        ExecutorService pool = Executors.newFixedThreadPool(parties);
        for (int i = 0; i < parties; i++) {
            final int idx = i;
            pool.submit(() -> {
                try {
                    for (int phase = 0; phase < 2; phase++) {   // 跑两轮，验证可重用
                        int s = 0;
                        for (int v : matrix[idx]) s += v + phase;
                        partial[idx] = s;
                        System.out.println(Thread.currentThread().getName()
                                + " 第 " + phase + " 阶段完成，部分和=" + s);
                        barrier.await();    // 到齐才继续；超时可写 await(5, TimeUnit.SECONDS)
                    }
                } catch (InterruptedException | BrokenBarrierException e) {
                    // BrokenBarrierException：有线程被中断/超时，栅栏被打破
                    System.err.println("栅栏被打破：" + e.getMessage());
                }
            });
        }
        pool.shutdown();
    }
}

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

/** 基于 Semaphore 的接口限流器：QPS 上限 + 超时快速失败 + 实时统计 */
public class SemaphoreLimiter {
    /** 同时允许的并发数（信号量许可数） */
    private final Semaphore semaphore;
    /** 是否公平：true 则先进先出，防止请求饥饿 */
    private final boolean fair;

    public SemaphoreLimiter(int permits, boolean fair) {
        this.semaphore = new Semaphore(permits, fair);
        this.fair = fair;
    }

    /** 阻塞式执行：拿不到许可就等 */
    public <T> T execute(Callable<T> task) throws Exception {
        semaphore.acquire();
        try {
            return task.call();
        } finally {
            semaphore.release();      // 必须 finally 归还，否则许可永久泄漏
        }
    }

    /** 限时执行：拿不到就快速失败，避免雪崩 */
    public <T> T execute(Callable<T> task, long timeout, TimeUnit unit) throws Exception {
        if (!semaphore.tryAcquire(timeout, unit)) {
            throw new RejectedExecutionException("系统繁忙，请稍后再试");
        }
        try {
            return task.call();
        } finally {
            semaphore.release();
        }
    }

    public int availablePermits() { return semaphore.availablePermits(); }
    public int queueLength()      { return semaphore.getQueueLength(); }

    public static void main(String[] args) throws InterruptedException {
        SemaphoreLimiter limiter = new SemaphoreLimiter(5, true);   // 最多 5 并发
        AtomicInteger ok = new AtomicInteger(), rejected = new AtomicInteger();
        CountDownLatch latch = new CountDownLatch(50);
        ExecutorService pool = Executors.newFixedThreadPool(50);

        for (int i = 0; i < 50; i++) {
            final int req = i;
            pool.submit(() -> {
                try {
                    limiter.execute(() -> {
                        TimeUnit.MILLISECONDS.sleep(200);   // 模拟耗时业务
                        return "req-" + req + " done";
                    }, 1, TimeUnit.SECONDS);                // 最多等 1 秒
                    ok.incrementAndGet();
                } catch (RejectedExecutionException e) {
                    rejected.incrementAndGet();             // 被限流
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    latch.countDown();
                }
            });
        }
        latch.await();
        System.out.println("成功 = " + ok.get() + "，被限流 = " + rejected.get());
        System.out.println("剩余许可 = " + limiter.availablePermits()
                + "，排队线程 = " + limiter.queueLength());
        pool.shutdown();
    }
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/** ThreadLocal 基本用法：用户上下文隐式传递 */
public class ThreadLocalBasic {
    private static final ThreadLocal<UserContext> CONTEXT = new ThreadLocal<>();

    static class UserContext {
        final String userId;
        UserContext(String userId) { this.userId = userId; }
    }

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        for (int i = 1; i <= 4; i++) {
            final String uid = "user-" + i;
            pool.submit(() -> {
                try {
                    CONTEXT.set(new UserContext(uid));   // 入口处设置
                    service();                            // 业务方法无需传参
                } finally {
                    CONTEXT.remove();                     // ★ 必须清理！线程池复用会串号
                }
            });
        }
        pool.shutdown();
    }

    static void service() { dao(); }
    static void dao() {
        UserContext ctx = CONTEXT.get();
        System.out.println(Thread.currentThread().getName() + " 执行 SQL，uid=" + ctx.userId);
    }
}

static class ThreadLocalMap {
    // ★ 关键：Entry 的 key 是弱引用
    static class Entry extends WeakReference<ThreadLocal<?>> {
        Object value;              // value 是强引用！
        Entry(ThreadLocal<?> k, Object v) {
            super(k);              // key 走 WeakReference
            value = v;
        }
    }
    private Entry[] table;
}

Thread(长生命周期) → ThreadLocalMap → Entry(key=null) → value(强引用，泄漏)

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 三个线程轮流打印 A/B/C 各 10 轮
 * 思路：一个 state 变量表示"该谁打印"，每个线程一个 Condition 精准唤醒下一个
 */
public class PrintABC {
    private static final ReentrantLock lock = new ReentrantLock();
    private static final Condition cA = lock.newCondition();
    private static final Condition cB = lock.newCondition();
    private static final Condition cC = lock.newCondition();
    private static int state = 0;          // 0->打印A, 1->打印B, 2->打印C
    private static final int ROUND = 10;

    static class Printer implements Runnable {
        private final String name;
        private final int target;                    // 我负责的 state
        private final Condition self;                // 我等待的条件
        private final Condition next;                // 我要唤醒的条件
        private final int nextState;

        Printer(String name, int target, Condition self, Condition next, int nextState) {
            this.name = name; this.target = target;
            this.self = self; this.next = next; this.nextState = nextState;
        }

        @Override
        public void run() {
            for (int i = 0; i < ROUND; i++) {
                lock.lock();
                try {
                    while (state != target)          // while 防虚假唤醒
                        self.await();
                    System.out.print(name);
                    state = nextState;
                    next.signal();                   // 精准唤醒下一个，不用 signalAll
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                } finally {
                    lock.unlock();
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Thread a = new Thread(new Printer("A", 0, cA, cB, 1), "T-A");
        Thread b = new Thread(new Printer("B", 1, cB, cC, 2), "T-B");
        Thread c = new Thread(new Printer("C", 2, cC, cA, 0), "T-C");
        a.start(); b.start(); c.start();
        a.join(); b.join(); c.join();
        System.out.println("\n完成");       // 输出：ABCABCABC...（共 10 轮）
    }
}

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

/** 用 Semaphore + BlockingQueue 实现一个简易数据库连接池 */
public class SimpleConnectionPool {
    private final BlockingQueue<Connection> pool;
    private final Semaphore permits;                 // 与池中连接数一致
    private final AtomicInteger created = new AtomicInteger();

    static class Connection {                        // 模拟连接
        final int id;
        Connection(int id) { this.id = id; }
        @Override public String toString() { return "Conn#" + id; }
    }

    public SimpleConnectionPool(int size) {
        this.pool = new ArrayBlockingQueue<>(size);
        this.permits = new Semaphore(size, true);    // 公平：先到先得
        for (int i = 0; i < size; i++) pool.add(new Connection(i));
    }

    /** 借连接：最多等 timeout；拿到许可后再 take（此时队列必有货） */
    public Connection borrow(long timeout, TimeUnit unit) throws InterruptedException {
        if (!permits.tryAcquire(timeout, unit))
            throw new IllegalStateException("连接池耗尽，等待超时");
        try {
            return pool.take();
        } catch (InterruptedException e) {
            permits.release();                       // 被打断要归还许可，避免泄漏
            throw e;
        }
    }

    /** 还连接：先放回队列再释放许可 */
    public void release(Connection c) {
        if (c == null) return;
        pool.offer(c);
        permits.release();
    }

    public int available() { return permits.availablePermits(); }

    public static void main(String[] args) throws InterruptedException {
        SimpleConnectionPool p = new SimpleConnectionPool(3);
        ExecutorService es = Executors.newFixedThreadPool(10);
        CountDownLatch latch = new CountDownLatch(10);
        for (int i = 0; i < 10; i++) {
            es.submit(() -> {
                Connection c = null;
                try {
                    c = p.borrow(1, TimeUnit.SECONDS);
                    System.out.println(Thread.currentThread().getName() + " 借到 " + c);
                    TimeUnit.MILLISECONDS.sleep(300);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } catch (IllegalStateException e) {
                    System.out.println(Thread.currentThread().getName() + " -> " + e.getMessage());
                } finally {
                    if (c != null) p.release(c);
                    latch.countDown();
                }
            });
        }
        latch.await();
        System.out.println("归还后可用连接 = " + p.available());   // 回到 3
        es.shutdown();
    }
}