LongAdder ⭐⭐

// 假设 128 个线程同时对同一个 AtomicLong 做累加
AtomicLong counter = new AtomicLong(0);
 
// 每个线程执行：
counter.incrementAndGet();
// 底层实际执行的是一个 CAS 循环：
// do {
//     oldVal = get();           // 读取当前值
//     newVal = oldVal + 1;      // 计算新值
// } while (!compareAndSet(oldVal, newVal));  // CAS 尝试写入，失败则重试

public class LongAdder extends Striped64 implements Serializable {
 
    // ========== 写操作 ==========
    public void add(long x);         // 累加指定值 x
    public void increment();          // 等价于 add(1L)
    public void decrement();          // 等价于 add(-1L)
 
    // ========== 读操作 ==========
    public long sum();                // 返回当前累计总和（非精确快照）
    public long longValue();          // 等价于 sum()，实现 Number 接口
    public int intValue();            // (int) sum()
    public float floatValue();        // (float) sum()
    public double doubleValue();      // (double) sum()
 
    // ========== 重置操作 ==========
    public void reset();              // 将 base 和所有 Cell 归零
    public long sumThenReset();       // 先求和再归零（非原子组合操作）
 
    // ========== 其他 ==========
    public String toString();         // 返回 sum() 的字符串形式
}

public class LongAdder extends Striped64 implements Serializable {
 
    /**
     * 创建一个初始值为 0 的 LongAdder。
     * 注意：此时 cells 数组为 null，base 为 0。
     * Cell 数组只有在第一次发生竞争时才会被懒初始化（lazy initialization）。
     */
    public LongAdder() {
        // 空构造函数，所有字段使用默认值
        // base = 0L (继承自 Striped64)
        // cells = null (继承自 Striped64)
        // cellsBusy = 0 (继承自 Striped64)
    }
}

// Striped64.java（LongAdder 的父类）
abstract class Striped64 extends Number {
    // base 值：在无竞争时直接累加到这里
    transient volatile long base;
 
    // Cell 数组：在出现竞争后才会被惰性初始化
    transient volatile Cell[] cells;
 
    // 自旋锁标志位：用于 cells 数组的初始化和扩容
    transient volatile int cellsBusy;
}

public void add(long x) {
    Cell[] cs;  // 局部变量引用 cells 数组
    long b;     // 局部变量保存 base 当前值
 
    // 第一层判断：如果 cells 为 null（还没初始化），尝试直接 CAS base
    if ((cs = cells) == null) {
        // casBase 底层是 Unsafe.compareAndSwapLong(this, BASE_OFFSET, b, b + x)
        if (casBase(b = base, b + x)) {
            // CAS 成功，直接返回，不需要触碰任何 Cell
            return;
        }
    }
    // CAS base 失败 或 cells 已存在 → 说明有竞争，走 Cell 路径
    // ...后续逻辑
}

public long sum() {
    Cell[] cs = cells; // 获取 cells 数组的快照引用
    long sum = base;   // ✅ 以 base 作为初始累加值
    if (cs != null) {
        for (Cell c : cs) {      // 遍历所有 Cell
            if (c != null) {
                sum += c.value;  // 累加每个 Cell 的值
            }
        }
    }
    return sum; // 返回 base + 所有 Cell.value 的总和
}

// 获取当前线程的探针值
int h = getProbe();
 
// 用位与操作定位 Cell 索引（等价于 h % cells.length，但更快）
int index = h & (cells.length - 1);
 
// 尝试对 cells[index] 进行 CAS
Cell c = cells[index];
if (c != null && c.cas(c.value, c.value + x)) {
    // 成功，直接返回
    return;
}

// 伪代码：Cell CAS 失败后的处理策略
if (cellCasFailed) {
    if (cells.length < NCPU) {
        // 还有扩容余地 → 扩容
        expandCells();
    }
    // 无论是否扩容，都 rehash 探针值，换个槽位重试
    advanceProbe(h);
}

// cellsBusy 充当自旋锁：0 表示空闲，1 表示被某线程持有
transient volatile int cellsBusy;
 
// 尝试获取锁：CAS 将 cellsBusy 从 0 改为 1
final boolean casCellsBusy() {
    return CELLSBUSY.compareAndSet(this, 0, 1);
}
 
// 扩容或初始化的核心逻辑（简化版）
if (casCellsBusy()) {         // 尝试获取锁
    try {
        if (cells == cs) {    // 双重检查（double-check），防止其他线程已经扩容
            Cell[] rs = new Cell[cs.length << 1];   // 容量翻倍
            for (int i = 0; i < cs.length; i++) {
                rs[i] = cs[i];                      // 迁移旧 Cell 引用
            }
            cells = rs;       // 发布新数组（volatile 写，确保可见性）
        }
    } finally {
        cellsBusy = 0;        // 释放锁（普通写即可，因为 volatile 保证了顺序）
    }
}

Cell 数组长度 = 4, CPU 核心数 = 8
 
Thread-A: probe = 0x3F → index = 3
Thread-B: probe = 0x7F → index = 3  (碰撞！)
Thread-C: probe = 0x42 → index = 2
Thread-D: probe = 0x10 → index = 0

Thread-B: advanceProbe(0x7F) → 新 probe = 0x5A → index = 2 或 index = 1

假设 N 个线程同时执行 add()，Cell 数组长度为 M
 
AtomicLong（M=1）：
  - 任意一次 CAS 的竞争概率 ≈ (N-1)/N
  - 16 线程时：竞争概率 ≈ 93.75%
 
LongAdder（M = min(N, NCPU)，理想均匀分布）：
  - 每个 Cell 上平均 N/M 个线程
  - 单个 Cell 的竞争概率 ≈ (N/M - 1) / (N/M)
  - 16 线程, 8 Cell：每 Cell 2 线程，竞争概率 ≈ 50%
  - 16 线程, 16 Cell：每 Cell 1 线程，竞争概率 ≈ 0%

// Striped64.java
final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) {
    int h;
    // 如果当前线程的探针值还没初始化（== 0），先初始化
    if ((h = getProbe()) == 0) {
        ThreadLocalRandom.current();   // 强制初始化探针值
        h = getProbe();                // 重新获取
        wasUncontended = true;         // 重置竞争标志
    }
 
    boolean collide = false;           // 标记：上一次是否发生了槽位碰撞
 
    for (;;) {                         // 无限重试循环（自旋）
        Cell[] cs; Cell c; int n; long v;
 
        // ▶ 分支一：cells 数组已存在
        if ((cs = cells) != null && (n = cs.length) > 0) {
 
            // 1) 目标槽位为空 → 尝试创建新 Cell 并放入
            if ((c = cs[(n - 1) & h]) == null) {
                if (cellsBusy == 0) {
                    Cell r = new Cell(x);           // 预先创建 Cell
                    if (cellsBusy == 0 && casCellsBusy()) {  // 获取自旋锁
                        try {
                            // 双重检查后安装 Cell
                            Cell[] rs; int m, j;
                            if ((rs = cells) != null &&
                                (m = rs.length) > 0 &&
                                rs[j = (m - 1) & h] == null) {
                                rs[j] = r;          // 安装成功
                                break;              // 退出自旋
                            }
                        } finally {
                            cellsBusy = 0;          // 释放锁
                        }
                    }
                }
                collide = false;                    // 安装失败，重置碰撞标志
 
            // 2) 上一次 CAS 失败且是已知的竞争 → 先标记，走到底部 rehash
            } else if (!wasUncontended) {
                wasUncontended = true;              // 标记已知，下轮重试
 
            // 3) 尝试 CAS 当前 Cell → 核心累加操作
            } else if (c.cas(v = c.value,
                             (fn == null) ? v + x : fn.applyAsLong(v, x))) {
                break;                              // CAS 成功，退出
 
            // 4) 数组长度已达 NCPU 上限 或 cells 已被其他线程替换 → 放弃扩容念头
            } else if (n >= NCPU || cells != cs) {
                collide = false;                    // 不扩容，只 rehash
 
            // 5) 首次碰撞 → 标记 collide，下轮再决定是否扩容
            } else if (!collide) {
                collide = true;
 
            // 6) 二次碰撞确认 → 执行扩容
            } else if (cellsBusy == 0 && casCellsBusy()) {
                try {
                    if (cells == cs) {
                        cells = Arrays.copyOf(cs, n << 1);  // 容量翻倍
                    }
                } finally {
                    cellsBusy = 0;
                }
                collide = false;
                continue;                           // 扩容后直接重试，不 rehash
            }
            h = advanceProbe(h);                    // 🔄 rehash 探针值
 
        // ▶ 分支二：cells 还不存在 → 初始化
        } else if (cellsBusy == 0 && cells == cs && casCellsBusy()) {
            try {
                if (cells == cs) {
                    Cell[] rs = new Cell[2];        // 初始容量为 2
                    rs[h & 1] = new Cell(x);        // 把当前值放入对应槽位
                    cells = rs;                     // 发布数组
                }
            } finally {
                cellsBusy = 0;
            }
            break;                                  // 初始化完成，退出
 
        // ▶ 分支三：初始化被其他线程抢了 → 退而求其次，再试一次 base CAS
        } else if (casBase(v = base,
                           (fn == null) ? v + x : fn.applyAsLong(v, x))) {
            break;                                  // base CAS 成功，退出
        }
    }
}

// Cell 是 Striped64 的静态内部类
// 使用 @jdk.internal.vm.annotation.Contended 注解修饰
@jdk.internal.vm.annotation.Contended  // 关键：避免伪共享（False Sharing）
static final class Cell {
    // volatile 保证可见性，存储该 Cell 槽位的当前累加值
    volatile long value;
 
    // 构造函数：初始化时传入一个初始值
    Cell(long x) {
        value = x;
    }
 
    // 通过 CAS 操作原子地更新 value
    // 参数 cmp 是期望值，val 是新值
    // 底层委托给 VarHandle 执行 compareAndSet
    final boolean cas(long cmp, long val) {
        // VALUE 是一个 VarHandle，指向 this.value 字段
        return VALUE.compareAndSet(this, cmp, val);
    }
 
    // 重置 value 为 identity（用于 Striped64 内部的表扩容等场景）
    final void reset(long identity) {
        // VALUE 是一个 VarHandle，此处使用 volatile 写语义
        VALUE.setVolatile(this, identity);
    }
 
    // 通过 VarHandle 获取对 value 字段的原子操作能力
    // 这是 JDK 9+ 取代 sun.misc.Unsafe 的现代写法
    private static final VarHandle VALUE;
 
    static {
        try {
            // 在类加载时，通过 MethodHandles.Lookup 查找 value 字段
            MethodHandles.Lookup l = MethodHandles.lookup();
            // 绑定 VarHandle 到 Cell 类的 value 字段
            VALUE = l.findVarHandle(Cell.class, "value", long.class);
        } catch (ReflectiveOperationException e) {
            // 如果反射查找失败，直接抛出错误（不应该发生）
            throw new ExceptionInInitializerError(e);
        }
    }
}

// ========== 未加 @Contended 的内存布局（伪共享风险） ==========
// |  Cell[0].value (8B)  |  Cell[1].value (8B)  |  Cell[2].value (8B)  | ...
// |<------------- 同一条 64B Cache Line 内 ------------->|
// 多个 Cell 的 value 挤在同一条缓存行中，互相干扰！
 
// ========== 加了 @Contended 的内存布局（伪共享消除） ==========
// | padding 128B | Cell[0].value (8B) | padding 128B |
// | padding 128B | Cell[1].value (8B) | padding 128B |
// | padding 128B | Cell[2].value (8B) | padding 128B |
// 每个 Cell 独占缓存行，彼此隔离，再无互相失效问题！

// Cell 的 CAS 更新方法
final boolean cas(long cmp, long val) {
    // cmp: 期望的当前值 (Compare value)
    // val: 要设置的新值 (New value)
    // 如果 this.value == cmp，则原子地将 this.value 设置为 val，返回 true
    // 如果 this.value != cmp（说明有其他线程已经修改过），返回 false
    return VALUE.compareAndSet(this, cmp, val);
}

// 伪代码：线程对 Cell 执行 +1 操作
Cell cell = cells[threadHash & (cells.length - 1)]; // 1. 通过哈希定位到某个 Cell
long currentValue = cell.value;                       // 2. 读取当前 volatile 值
boolean success = cell.cas(currentValue, currentValue + 1); // 3. CAS 尝试写入新值
// 如果 success == true，累加成功，直接返回
// 如果 success == false，说明有竞争，进入 Striped64.longAccumulate() 处理

private static final VarHandle VALUE;  // 声明为 static final，全局唯一
 
static {
    try {
        // MethodHandles.lookup() 获取当前类的 Lookup 对象
        // 因为 Cell 是 Striped64 的内部类，有权限访问自己的 private 字段
        MethodHandles.Lookup l = MethodHandles.lookup();
        // findVarHandle 三个参数：
        //   1. Cell.class —— 目标类
        //   2. "value"    —— 目标字段名
        //   3. long.class —— 目标字段类型
        VALUE = l.findVarHandle(Cell.class, "value", long.class);
    } catch (ReflectiveOperationException e) {
        // 这个异常理论上永远不会发生（因为 value 字段确实存在）
        // 但如果发生了，说明 JVM 内部出现严重问题，直接抛出错误终止
        throw new ExceptionInInitializerError(e);
    }
}

// ╔══════════════════════════════════════════════════════════╗
// ║              Cell 对象在堆内存中的实际布局                ║
// ╠══════════════════════════════════════════════════════════╣
// ║  [Object Header]           12 ~ 16 bytes                ║
// ║  ──────────────────────────────────────────              ║
// ║  [Contended Padding前]     128 bytes (全零填充)          ║
// ║  ──────────────────────────────────────────              ║
// ║  [volatile long value]     8 bytes   ← 唯一有效数据      ║
// ║  ──────────────────────────────────────────              ║
// ║  [Contended Padding后]     128 bytes (全零填充)          ║
// ╠══════════════════════════════════════════════════════════╣
// ║  总计约：~280 bytes / 每个 Cell 对象                     ║
// ║  其中有效数据仅 8 bytes，其余全是为消除伪共享的填充！      ║
// ╚══════════════════════════════════════════════════════════╝

// LongAdder.java — 核心累加方法
public void add(long x) {
    Cell[] cs;    // cs: 本地引用，指向 cells 数组
    long b, v;    // b: base 值; v: 某个 Cell 的当前值
    int m;        // m: cells 数组长度减 1，用于取模定位
    Cell c;       // c: 当前线程映射到的 Cell 对象
 
    // 【第一层判断】：尝试直接 CAS 更新 base
    // 如果 cells 数组尚未初始化（cs == null），就尝试对 base 做 CAS
    // 如果 CAS 成功，说明竞争不激烈，直接返回——这是最快路径 (fast path)
    if ((cs = cells) != null ||               // 条件 A：cells 已存在，说明曾发生过竞争
        !casBase(b = base, b + x)) {          // 条件 B：CAS base 失败，说明正在发生竞争
 
        // 走到这里，意味着两种情况之一：
        // 1. cells 数组已经存在（之前已发生过竞争）
        // 2. cells 不存在，但 CAS base 失败了（此刻正在竞争）
 
        boolean uncontended = true;  // 标记当前线程对 Cell 的 CAS 是否成功（无竞争）
 
        // 【第二层判断】：尝试定位到某个 Cell 并 CAS 更新
        if (cs == null ||                                 // 条件 1：cells 尚未初始化
            (m = cs.length - 1) < 0 ||                    // 条件 2：cells 长度为 0（理论防御）
            (c = cs[getProbe() & m]) == null ||           // 条件 3：当前线程哈希到的槽位为 null
            !(uncontended = c.cas(v = c.value, v + x)))   // 条件 4：对该 Cell CAS 失败
        {
            // 以上任一条件成立，都进入 longAccumulate——重量级处理逻辑
            longAccumulate(x, null, uncontended);
        }
    }
}

// 当 cells 数组未初始化，且 CAS base 成功时，直接返回
if ((cs = cells) != null || !casBase(b = base, b + x)) {
    // ... 进入竞争处理
}

// getProbe() 获取当前线程的哈希探针值
// 通过 & m（等价于 % length）定位到 cells 数组中的某个槽位
(c = cs[getProbe() & m]) == null ||
!(uncontended = c.cas(v = c.value, v + x))

// Striped64.java — 重量级累加逻辑（简化骨架）
final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) {
    int h;  // h: 当前线程的探针哈希值
    // 如果探针值为 0，说明线程尚未初始化哈希，先强制初始化
    if ((h = getProbe()) == 0) {
        ThreadLocalRandom.current();   // 强制初始化 threadLocalRandomProbe
        h = getProbe();                // 重新获取探针值
        wasUncontended = true;         // 重新初始化后视为"无竞争"，给一次重试机会
    }
 
    boolean collide = false;  // 标记是否发生了槽位碰撞（用于决定是否扩容）
 
    // 自旋循环——直到成功为止
    for (;;) {
        Cell[] cs; Cell c; int n; long v;
 
        // ————— 分支一：cells 已初始化 —————
        if ((cs = cells) != null && (n = cs.length) > 0) {
 
            if ((c = cs[(n - 1) & h]) == null) {
                // 【场景 A】：目标槽位为空 → 创建新 Cell 并放入
                // 使用 cellsBusy 自旋锁保护数组写操作
                // 创建 Cell(x)，通过 CAS 将 cellsBusy 设为 1 获取锁
                // 成功后将新 Cell 放入数组，释放锁，跳出循环
                // ...（省略具体实现）
            }
            else if (!wasUncontended) {
                // 【场景 B】：上一次 CAS 失败（从 add 方法传入 uncontended=false）
                // 不立即重试同一个槽位，而是先 rehash，换个槽位再试
                wasUncontended = true;    // 重置标记
                // 继续往下执行 advanceProbe(h) 换槽位
            }
            else if (c.cas(v = c.value, (fn == null) ? v + x : fn.applyAsLong(v, x))) {
                // 【场景 C】：对当前 Cell CAS 成功 → 直接跳出循环
                break;
            }
            else if (n >= NCPU || cells != cs) {
                // 【场景 D】：数组长度已达 CPU 核心数上限，或数组引用已变化
                // 不再扩容，只做 rehash 换槽位
                collide = false;
            }
            else if (!collide) {
                // 【场景 E】：首次碰撞，标记 collide=true，下次循环如果还失败就扩容
                collide = true;
            }
            else if (cellsBusy == 0 && casCellsBusy()) {
                // 【场景 F】：连续碰撞 → 扩容！容量翻倍
                // 将 cells 数组长度扩大为原来的 2 倍
                // 扩容后释放锁
                collide = false;
                continue;       // 扩容后重新循环，不做 rehash
            }
 
            h = advanceProbe(h);  // rehash：更换当前线程的探针值，下次映射到不同槽位
        }
 
        // ————— 分支二：cells 尚未初始化 → 初始化数组 —————
        else if (cellsBusy == 0 && cells == cs && casCellsBusy()) {
            // 获取 cellsBusy 锁成功
            // 创建初始长度为 2 的 Cell 数组
            // 将当前值 x 放入对应槽位的新 Cell 中
            // 释放锁，跳出循环
            // ...（省略具体实现）
        }
 
        // ————— 分支三：初始化数组时锁竞争失败 → 退回尝试 CAS base —————
        else if (casBase(v = base, (fn == null) ? v + x : fn.applyAsLong(v, x))) {
            break;  // 退而求其次，CAS base 成功就返回
        }
    }
}

// Striped64.java — 探针值推进（xorshift 算法）
static final int advanceProbe(int probe) {
    probe ^= probe << 13;   // xorshift 第一步：左移异或
    probe ^= probe >>> 17;  // xorshift 第二步：无符号右移异或
    probe ^= probe << 5;    // xorshift 第三步：左移异或
    THREAD_PROBE.set(Thread.currentThread(), probe);  // 写回线程的探针字段
    return probe;            // 返回新的探针值
}

// 当 cells 长度 >= CPU 数量时，不再扩容
else if (n >= NCPU || cells != cs) {
    collide = false;  // 放弃扩容念头，只做 rehash
}

// 第一次碰撞：只标记 collide = true，不扩容
else if (!collide) {
    collide = true;
}
// 第二次碰撞：真正执行扩容
else if (cellsBusy == 0 && casCellsBusy()) {
    // 扩容为原来的 2 倍
}

// cellsBusy 充当轻量级自旋锁
// 0 = 未锁定; 1 = 已锁定
volatile int cellsBusy;
 
final boolean casCellsBusy() {
    return CELLSBUSY.compareAndSet(this, 0, 1);  // CAS 将 0 设为 1
}

// LongAdder.java —— JDK 源码
public long sum() {
    // 1. 获取 Cell 数组的引用（注意：这不是一个原子快照）
    Cell[] cs = cells;
 
    // 2. 以 base 值作为累加的起点
    //    base 承载了无竞争时的所有写入，以及竞争初期尚未分流的写入
    long sum = base;
 
    // 3. 如果 Cell 数组已经被初始化（说明曾经发生过竞争）
    if (cs != null) {
        // 4. 遍历 Cell 数组中的每一个槽位
        for (Cell c : cs) {
            // 5. 跳过尚未被使用的空槽位（Cell 是惰性初始化的）
            if (c != null) {
                // 6. 将该 Cell 的 value 累加到 sum 上
                //    注意：这里直接读取 c.value，没有任何加锁或 CAS
                sum += c.value;
            }
        }
    }
 
    // 7. 返回 base + 所有非空 Cell 的 value 之和
    return sum;
}

// Striped64.java（LongAdder 的父类）
transient volatile long base;          // volatile，保证单次读的可见性
 
// Cell 内部
@jdk.internal.vm.annotation.Contended
static final class Cell {
    volatile long value;               // volatile，保证单次读的可见性
    // ...
}

// LongAdder.java
public long sumThenReset() {
    Cell[] cs = cells;
    // 读取并重置 base（注意：这也不是原子操作！）
    long sum = getAndSetBase(0);  // 原子地读取 base 并置零
    if (cs != null) {
        for (Cell c : cs) {
            if (c != null) {
                // 原子地读取 cell.value 并置零
                sum += c.getAndSet(0);
            }
        }
    }
    return sum;
}

// 将 base 和所有 Cell 重置为 0
// 注意：如果存在并发写入，重置后的值不保证为 0
public void reset() {
    Cell[] cs = cells;
    // 将 base 重置为 0
    base = 0L;
    if (cs != null) {
        // 遍历所有 Cell，逐个重置
        for (Cell c : cs) {
            if (c != null) {
                c.reset();  // 内部将 value 置为 0
            }
        }
    }
}

// === AtomicLong.incrementAndGet() 的内部实现 (简化) ===
public final long incrementAndGet() {
    // unsafe.getAndAddLong 底层是一个 CAS 循环
    // 所有线程竞争同一个 value 地址
    return U.getAndAddLong(this, VALUE, 1L) + 1L;
}
 
// === 等价的 CAS 循环展开 ===
public final long getAndAddLong(Object o, long offset, long delta) {
    long v;                                    // 用于存储当前读到的旧值
    do {
        v = getLongVolatile(o, offset);        // 每次循环都要重新读取最新值 (volatile read)
    } while (!weakCompareAndSetLong(           // CAS 尝试写入: 期望值 v, 新值 v+delta
            o, offset, v, v + delta));         // 失败则继续自旋, 高并发下这里会空转大量 CPU 周期
    return v;                                  // 返回旧值
}

┌─────────────────────── Cache Line (64 bytes) ────────────────────────┐
│                                                                      │
│  AtomicLong: [ value(8B) | padding... ]                              │
│  所有核心竞争这一条 cache line  ──→  cache line bouncing             │
│                                                                      │
├──────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  LongAdder Cell[0]: [ @Contended padding(128B) | value(8B) | pad ]  │
│  LongAdder Cell[1]: [ @Contended padding(128B) | value(8B) | pad ]  │
│  LongAdder Cell[2]: [ @Contended padding(128B) | value(8B) | pad ]  │
│  各核心各写各的 cache line  ──→  几乎零缓存争用                      │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.CountDownLatch;
 
public class AdderVsAtomicBenchmark {
 
    // 每个线程的累加次数
    private static final int OPS_PER_THREAD = 10_000_000;
 
    public static void main(String[] args) throws Exception {
        // 分别测试不同线程数下的耗时
        int[] threadCounts = {1, 2, 4, 8, 16, 32, 64};
 
        for (int nThreads : threadCounts) {
            long atomicTime = benchmarkAtomicLong(nThreads);     // 测试 AtomicLong
            long adderTime = benchmarkLongAdder(nThreads);       // 测试 LongAdder
            // 输出对比结果
            System.out.printf("Threads=%2d | AtomicLong=%5dms | LongAdder=%5dms | Speedup=%.2fx%n",
                    nThreads, atomicTime, adderTime,             // 打印耗时
                    (double) atomicTime / adderTime);            // 计算加速比
        }
    }
 
    // AtomicLong 基准测试
    static long benchmarkAtomicLong(int nThreads) throws Exception {
        AtomicLong counter = new AtomicLong(0);                  // 共享的原子计数器
        CountDownLatch latch = new CountDownLatch(nThreads);     // 等待所有线程完成的门闩
 
        long start = System.currentTimeMillis();                 // 记录开始时间
        for (int i = 0; i < nThreads; i++) {
            new Thread(() -> {
                for (int j = 0; j < OPS_PER_THREAD; j++) {
                    counter.incrementAndGet();                   // 所有线程竞争同一个 value
                }
                latch.countDown();                               // 当前线程完成, 门闩减一
            }).start();
        }
        latch.await();                                           // 等待所有线程结束
        return System.currentTimeMillis() - start;               // 返回总耗时
    }
 
    // LongAdder 基准测试
    static long benchmarkLongAdder(int nThreads) throws Exception {
        LongAdder adder = new LongAdder();                       // 共享的分段累加器
        CountDownLatch latch = new CountDownLatch(nThreads);     // 同步门闩
 
        long start = System.currentTimeMillis();                 // 记录开始时间
        for (int i = 0; i < nThreads; i++) {
            new Thread(() -> {
                for (int j = 0; j < OPS_PER_THREAD; j++) {
                    adder.increment();                           // 线程被分散到不同 Cell, 竞争大幅降低
                }
                latch.countDown();                               // 当前线程完成
            }).start();
        }
        latch.await();                                           // 等待所有线程结束
        return System.currentTimeMillis() - start;               // 返回总耗时
    }
}

public long sum() {
    Cell[] cs = cells;                      // 读取 cells 数组引用 (非 volatile 语义下可能读到旧值)
    long sum = base;                        // 先累加 base
    if (cs != null) {                       // 如果存在 Cell 数组
        for (Cell c : cs) {                 // 遍历每一个 Cell
            if (c != null) {
                sum += c.value;             // 累加每个 Cell 的 value
            }
            // ⚠️ 关键问题: 在遍历过程中, 其他线程仍然在修改 base 和各 Cell 的 value
            // 已经遍历过的 Cell 可能在此刻又被其他线程增加了
            // 尚未遍历的 Cell 可能在被遍历之前又被修改
            // 因此 sum 只是一个 "近似快照", 而非精确的原子快照
        }
    }
    return sum;                             // 返回的是一个弱一致性的近似值
}

AtomicLong counter = new AtomicLong(0);
 
// 1. 原子性的 "读-改-写" —— 返回精确的新值
long precise = counter.incrementAndGet();       // 返回值一定是全局唯一的, 可用于生成序列号
 
// 2. 原子性的条件更新 —— CAS 语义
boolean success = counter.compareAndSet(        // 只有当前值等于 expected 时才更新
        100L,                                   // expected: 期望的当前值
        200L                                    // update: 要设置的新值
);
 
// 3. 原子性的获取并更新 —— 函数式 API (Java 8+)
long oldVal = counter.getAndUpdate(             // 传入一个函数, 原子地应用到当前值
        current -> current * 2                  // 将当前值翻倍, 返回旧值
);
 
// 4. 原子性的累加并获取
long newVal = counter.accumulateAndGet(         // 将当前值与参数通过函数合并
        5L,                                     // 参与运算的外部值
        Long::max                               // 取当前值和5的较大者
);

// ❌ 错误用法: 用 LongAdder 做限流
public class BrokenRateLimiter {
    private final LongAdder permits = new LongAdder();  // 已消耗的许可数
    private final long maxPermits = 1000;               // 最大许可数
 
    public boolean tryAcquire() {
        // 问题1: sum() 不精确, 可能同时有多个线程看到 "还有余额"
        if (permits.sum() < maxPermits) {
            // 问题2: increment() 和上面的 sum() 之间不是原子操作
            // 可能超发许可, 导致实际消耗 > maxPermits
            permits.increment();
            return true;
        }
        return false;
    }
}
 
// ✅ 正确用法: 用 AtomicLong 做限流
public class CorrectRateLimiter {
    private final AtomicLong permits = new AtomicLong(0); // 已消耗的许可数
    private final long maxPermits = 1000;                 // 最大许可数
 
    public boolean tryAcquire() {
        long current;                                     // 当前已消耗数
        do {
            current = permits.get();                      // 读取精确的当前值
            if (current >= maxPermits) {                  // 如果已经到达上限
                return false;                             // 直接拒绝, 无需 CAS
            }
        } while (!permits.compareAndSet(                  // CAS 原子地将 current 更新为 current+1
                current, current + 1));                   // 失败则重试, 保证不会超发
        return true;                                      // CAS 成功, 许可获取成功
    }
}

// LongAccumulator 的构造函数签名
public LongAccumulator(
    LongBinaryOperator accumulatorFunction, // 累加函数：定义两个 long 值如何合并
    long identity                            // 初始值（幺元）：满足 f(identity, x) == x
) {
    this.function = accumulatorFunction;     // 保存累加函数引用
    base = this.identity = identity;         // base 初始化为 identity
}

// accumulate 方法的核心逻辑（简化版，展示关键差异）
public void accumulate(long x) {
    Cell[] cs;                              // 本地引用 Cell 数组
    long b, v, r;                           // b=base值, v=Cell值, r=函数计算结果
    int m;                                  // Cell 数组长度掩码
    Cell c;                                 // 当前线程命中的 Cell
 
    // 第一步：尝试直接在 base 上进行累加（无竞争快路径）
    if ((cs = cells) != null ||             // 如果 Cell 数组已初始化（说明曾发生过竞争）
        ((r = function.applyAsLong(b = base, x)) != b  // 计算 f(base, x)
         && !casBase(b, r))) {              // 尝试 CAS 更新 base，失败说明有竞争
        // ——进入分段逻辑——
        boolean uncontended = true;         // 标记当前 Cell 是否无竞争
        if (cs == null ||                   // Cell 数组尚未初始化
            (m = cs.length - 1) < 0 ||      // Cell 数组为空
            (c = cs[getProbe() & m]) == null || // 当前线程哈希到的槽位为空
            !(uncontended =                 // 尝试 CAS 更新命中的 Cell
                (r = function.applyAsLong(v = c.value, x)) == v // 计算 f(cell.value, x)
                || c.cas(v, r))) {          // CAS 尝试写入
            longAccumulate(x, function, uncontended); // 进入 Striped64 的通用扩容/重试逻辑
        }
    }
}

public long get() {
    Cell[] cs = cells;                      // 获取 Cell 数组引用
    long result = base;                     // 从 base 值开始累积
    if (cs != null) {                       // 如果存在 Cell 数组
        for (Cell c : cs) {                 // 遍历每一个 Cell 槽位
            if (c != null) {                // 跳过尚未初始化的空槽
                result = function.applyAsLong(result, c.value); // 用函数合并
            }
        }
    }
    return result;                          // 返回最终合并结果
}

// 场景一：并发求最大值
// 幺元为 Long.MIN_VALUE，因为 max(MIN_VALUE, x) == x
LongAccumulator maxAccumulator = new LongAccumulator(
    Long::max,               // (left, right) -> Math.max(left, right)
    Long.MIN_VALUE           // 任何值都比它大，不影响最终结果
);
 
// 场景二：并发求最小值
// 幺元为 Long.MAX_VALUE，因为 min(MAX_VALUE, x) == x
LongAccumulator minAccumulator = new LongAccumulator(
    Long::min,               // (left, right) -> Math.min(left, right)
    Long.MAX_VALUE           // 任何值都比它小，不影响最终结果
);
 
// 场景三：并发求累加（等价于 LongAdder）
// 幺元为 0，因为 0 + x == x
LongAccumulator sumAccumulator = new LongAccumulator(
    Long::sum,               // (left, right) -> left + right
    0L                       // 加法单位元
);
 
// 场景四：并发求按位或（收集所有标志位）
// 幺元为 0，因为 0 | x == x
LongAccumulator orAccumulator = new LongAccumulator(
    (left, right) -> left | right,  // 按位或合并
    0L                               // 按位或的单位元
);

public class LongAccumulatorMaxDemo {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个求最大值的累加器，初始值为 Long.MIN_VALUE
        LongAccumulator max = new LongAccumulator(Long::max, Long.MIN_VALUE);
 
        // 模拟 10 个线程，每个线程写入一批随机值
        int threadCount = 10;                           // 线程数量
        ExecutorService pool = Executors.newFixedThreadPool(threadCount); // 创建线程池
 
        ThreadLocalRandom random = ThreadLocalRandom.current(); // 线程安全的随机数生成器
 
        for (int i = 0; i < threadCount; i++) {         // 提交 10 个任务
            pool.submit(() -> {
                for (int j = 0; j < 100_000; j++) {     // 每个线程执行 10 万次写入
                    long value = ThreadLocalRandom.current().nextLong(1_000_000);
                    max.accumulate(value);               // 将随机值喂入累加器
                }
            });
        }
 
        pool.shutdown();                                 // 不再接受新任务
        pool.awaitTermination(10, TimeUnit.SECONDS);     // 等待所有任务完成
 
        // 所有线程完成后，get() 返回所有写入值中的最大值
        System.out.println("Max value: " + max.get());   // 输出最大值
    }
}

// 错误示范：减法不满足交换律和结合律！
LongAccumulator badAccumulator = new LongAccumulator(
    (left, right) -> left - right,   // 减法：不满足交换律 a-b != b-a
    0L                                // 看似合理的幺元
);

假设 3 个线程分别写入值 10, 20, 30
 
可能的合并顺序 1: ((0 - 10) - 20) - 30 = -60
可能的合并顺序 2: ((0 - 30) - 10) - 20 = -60  ← 碰巧相同？
可能的合并顺序 3:
  Cell[0] 收到 10, Cell[1] 收到 20, base 收到 30
  get() 合并: f(f(base, Cell[0]), Cell[1])
            = f(f(30, 10), 20)
            = f(30 - 10, 20)
            = 20 - 20
            = 0                                  ← 结果完全错误！
 
根本原因：分段机制将值分散到不同位置，
合并时的顺序和写入时的顺序无关，
减法因不满足交换律和结合律，导致结果不确定。

LongAccumulator acc = new LongAccumulator(Long::max, Long.MIN_VALUE);
 
// ... 经过一段时间的并发写入 ...
 
// getThenReset()：原子地获取当前值并重置为 identity
long windowMax = acc.getThenReset();     // 返回当前最大值，同时将 base 和所有 Cell 重置为 MIN_VALUE
System.out.println("本窗口最大值: " + windowMax);
 
// reset()：仅重置，不返回值
acc.reset();                              // 将 base 和所有 Cell 重置为 identity

// DoubleAccumulator 示例：并发求浮点数的最大值
DoubleAccumulator doubleMax = new DoubleAccumulator(
    Double::max,                          // 双精度浮点取最大值
    Double.NEGATIVE_INFINITY              // 负无穷作为幺元
);
 
doubleMax.accumulate(3.14);               // 写入值
doubleMax.accumulate(2.718);              // 写入值
System.out.println(doubleMax.get());      // 输出 3.14

LongAccumulator acc = new LongAccumulator((left, right) -> left * right, 1L);
acc.accumulate(3);
acc.accumulate(5);
acc.accumulate(2);
System.out.println(acc.get());