LongAdder
LongAdder 源码分析
AtomicLong 在高并发写入时, 所有线程挤在同一个 long 上 CAS 自旋, 失败重试白烧 CPU。
LongAdder 把”一个热点”拆成”base + 一张 Cell 表”: 没竞争时只写 base, 有竞争就按线程哈希散到不同 Cell 上各写各的, 读的时候再把 base 和所有 Cell 加起来。写扩展、读汇总, 用空间换吞吐。
代码块收起展开
// 基于 JDK 25 (本地 D:/1ForCode/JAVA_Source), java.util.concurrent.atomic.LongAdder
public class LongAdder extends Striped64 implements Serializable {
public void add(long x) {
Cell[] cs; long b, v; int m; Cell c;
if ((cs = cells) != null || !casBase(b = base, b + x)) { // 快路径: cells 未创建 且 base 一次 CAS 成功 -> 到此结束
int index = getProbe(); // 线程的 probe 哈希(存在 Thread 里), 决定落到哪个槽
boolean uncontended = true;
if (cs == null || (m = cs.length - 1) < 0 || // 表还没建
(c = cs[index & m]) == null || // 命中的槽还是空的
!(uncontended = c.cas(v = c.value, v + x))) // 槽上 CAS 一次, 失败说明这个槽也撞车了
longAccumulate(x, null, uncontended, index); // 任一条件不满足 -> 进 Striped64 慢路径兜底
}
}
public void increment() {
add(1L); // increment/decrement 只是 add 的别名, 没有独立实现
}
// ...
public long sum() {
Cell[] cs = cells;
long sum = base; // 无锁遍历累加: 遍历途中别的线程还在写, 所以只是"某一时刻附近"的快照
if (cs != null) {
for (Cell c : cs)
if (c != null)
sum += c.value;
}
return sum;
}
public void reset() {
Cell[] cs = cells;
base = 0L; // 逐个清零, 整个过程不是原子的, 只能在确认无并发写时用
if (cs != null) {
for (Cell c : cs)
if (c != null)
c.reset();
}
}
// ... sumThenReset / Number 的四个 xxxValue / 序列化代理省略, 序列化时只存 sum() 的结果
}真正的机制都在父类 Striped64 里, LongAdder/DoubleAdder/LongAccumulator 共用这一套:
代码块收起展开
// 基于 JDK 25 (本地 D:/1ForCode/JAVA_Source), java.util.concurrent.atomic.Striped64
abstract class Striped64 extends Number {
// Cell 就是一个只留了 CAS 原语的裸 AtomicLong。@Contended 让 JVM 给它前后填充缓存行:
// 散落在堆里的 AtomicLong 一般不挨着, 但数组里的 Cell 必然相邻, 不填充就会伪共享——
// 两个线程写不同 Cell, 却因为落在同一 64 字节缓存行上互相打飞对方的缓存, 分散就白做了
@jdk.internal.vm.annotation.Contended static final class Cell {
volatile long value;
Cell(long x) { value = x; }
final boolean cas(long cmp, long val) {
return VALUE.weakCompareAndSetRelease(this, cmp, val); // weak+release: 允许偶发假失败(外层本来就在循环重试), 换更便宜的内存屏障
}
// ... reset / getAndSet / VarHandle 声明省略
}
/** Number of CPUS, to place bound on table size */
static final int NCPU = Runtime.getRuntime().availableProcessors();
transient volatile Cell[] cells; // 懒初始化, 容量恒为 2 的幂, 索引用 (n-1) & probe
transient volatile long base; // 无竞争时的累加目标, 也是表初始化期间的退路
transient volatile int cellsBusy; // 自旋锁标志位(0/1), 保护建表/扩容/往槽里放新 Cell
final boolean casBase(long cmp, long val) {
return BASE.weakCompareAndSetRelease(this, cmp, val);
}
final boolean casCellsBusy() {
return CELLSBUSY.compareAndSet(this, 0, 1); // 拿"锁"就是把 0 CAS 成 1, 释放就是普通写回 0
}
static final int getProbe() {
return TLR.getThreadLocalRandomProbe(); // 复用 ThreadLocalRandom 的 probe 当线程哈希, 不必新增字段
}
static final int advanceProbe(int probe) {
return TLR.advanceThreadLocalRandomProbe(probe); // Marsaglia XorShift 推进, 换一个槽再试
}
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended, int index) {
if (index == 0) { // probe 为 0 = 线程还没初始化过 ThreadLocalRandom
ThreadLocalRandom.current(); // force initialization
index = getProbe();
wasUncontended = true; // 之前的失败是拿默认哈希 0 撞的, 不算数
}
for (boolean collide = false;;) { // True if last slot nonempty
Cell[] cs; Cell c; int n; long v;
if ((cs = cells) != null && (n = cs.length) > 0) {
if ((c = cs[(n - 1) & index]) == null) {
if (cellsBusy == 0) { // Try to attach new Cell
Cell r = new Cell(x); // Optimistically create // 锁外先建好对象, 缩短持锁时间
if (cellsBusy == 0 && casCellsBusy()) {
try { // Recheck under lock
Cell[] rs; int m, j;
if ((rs = cells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & index] == null) { // 锁内重查: 期间可能已被别人放上或扩容
rs[j] = r; // 值直接带进新 Cell, 本次累加完成
break;
}
} finally {
cellsBusy = 0; // 释放锁只需普通写: 只有持锁者会写这个字段
}
continue; // Slot is now non-empty
}
}
collide = false;
}
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash // 进来前已在这个槽失败过, 先换槽, 别原地再撞
else if (c.cas(v = c.value,
(fn == null) ? v + x : fn.applyAsLong(v, x)))
break; // fn==null 就是 LongAdder 的加法, 省掉一个函数对象字段
else if (n >= NCPU || cells != cs)
collide = false; // At max size or stale // 到容量上限或表已被换掉, 不再考虑扩容
else if (!collide)
collide = true; // 第一次真碰撞先记账, 换槽再撞一次才扩容
else if (cellsBusy == 0 && casCellsBusy()) {
try {
if (cells == cs) // Expand table unless stale
cells = Arrays.copyOf(cs, n << 1); // 翻倍。旧 Cell 原样搬过去, 不需要 rehash 数据
} finally {
cellsBusy = 0;
}
collide = false;
continue; // Retry with expanded table
}
index = advanceProbe(index); // 每轮失败都换哈希, 逃离热点槽
}
else if (cellsBusy == 0 && cells == cs && casCellsBusy()) {
try { // Initialize table
if (cells == cs) {
Cell[] rs = new Cell[2]; // 首次竞争才建表, 初始只有 2 个槽
rs[index & 1] = new Cell(x); // 建表和第一次写一步完成
cells = rs;
break;
}
} finally {
cellsBusy = 0;
}
}
// Fall back on using base
else if (casBase(v = base,
(fn == null) ? v + x : fn.applyAsLong(v, x)))
break; // 建表的锁被别人拿着, 那就退回去写 base, 绝不空等
}
}
// ... doubleAccumulate 是同一套逻辑加 long/double 位转换, 注释明说是 copy/paste/adapt 维护的
}原理串讲
一次 add(x) 的完整链路: 先看 cells 是否为 null。整个类还没经历过竞争时它就是 null, 于是直接 casBase(b, b+x), 一次 CAS 成功就返回, 这条快路径和 AtomicLong 几乎等价, 这也是”低竞争下两者性能相近”的原因。
快路径断掉有两种可能: base 的 CAS 失败(第一次撞车), 或者 cells 已经存在(历史上撞过车)。
此时取 getProbe() 拿线程哈希, 用 index & (length-1) 定位槽, 对槽里的 Cell 做一次 CAS。
这次也失败, 或者表/槽还不存在, 才掉进 longAccumulate。
longAccumulate 是一个大自旋, 每轮按当前状态走一个分支。表不存在: 抢 cellsBusy 建一张容量 2 的表, 顺手把 x 放进新 Cell, 一步完成; 抢不到锁就退化成 casBase, 绝不阻塞等待。
槽为空: 锁外先 new Cell(x), 抢到锁后重查槽仍为空才放进去——重查是因为从”看见空槽”到”拿到锁”之间, 别的线程可能已经放了 Cell 或换了表。
槽非空: CAS 累加, 失败则依次升级——先 advanceProbe 换哈希换槽(rehash), 换槽后还撞(collide 置位后再失败)才抢锁把表翻倍。
n >= NCPU 后永不扩容, 只靠不断 rehash 把线程摊开。
两处”为什么”值得记住。其一, 为什么扩容上限是 NCPU: 同时真正在写的线程最多就是 CPU 核数个, 槽比核多不会减少碰撞, 只浪费内存; 理论上存在一个”每核一槽”的完美哈希, rehash 就是在随机搜索它。
其二, 为什么 cellsBusy 用自旋标志而非真正的锁: 它保护的临界区只有几条赋值语句, 而且拿不到锁的线程有退路(换槽/写 base), park/unpark 一个线程的代价远超临界区本身。
同理 Cell.cas 用 weakCompareAndSetRelease: 调用点全在重试循环里, 假失败无非多转一圈, 换来的是省掉完整的 volatile 语义屏障。
sum() 没有任何同步, 就是 base 加所有 Cell 的当前值。遍历到第 3 个 Cell 时第 1 个又被加了, 这次加不会体现在结果里, 所以它是非原子快照: 没有并发写时精确, 有并发写时是”最终会对”的近似值。
这正是选型的分水岭——统计计数(QPS、命中数、监控埋点)要的是高吞吐写和最终准确, 用 LongAdder; 需要 incrementAndGet 的返回值参与逻辑、或要求任意时刻读到精确值(发号器、限流令牌), 只能用 AtomicLong, 因为 LongAdder 根本没有”加完后的全局值”这个概念。
设计取舍
- 空间换吞吐: 一个 @Contended Cell 连填充要上百字节, 所以全程懒创建——不撞车不建表, 表从 2 起步按需翻倍。
sum()弱一致不是缺陷而是定价: 想要精确快照就得全局锁或全局 CAS, 那分散写就白做了。increment()没有返回值, 注意: LongAdder 无法告诉你”你这一下之后总数是多少”, 需要该语义直接排除它。reset()/sumThenReset()天然 racy, 只适合任务批次之间的静止期调用, 并发写入下丢数没商量。- probe 复用 ThreadLocalRandom 的线程内字段, ConcurrentHashMap 的计数器(CounterCell)也是同一套 Striped64 思路。
附: AtomicReference
把 CAS 从 long 搬到对象引用上, 整个类就是一个 volatile 字段加一组 VarHandle 转发, 是无锁栈、无锁缓存刷新(不可变对象整体替换)的基础件。
代码块收起展开
// 基于 JDK 25 (本地 D:/1ForCode/JAVA_Source), java.util.concurrent.atomic.AtomicReference
public class AtomicReference<V> implements java.io.Serializable {
private static final VarHandle VALUE = MhUtil.findVarHandle(
MethodHandles.lookup(), "value", Object.class);
@SuppressWarnings("serial") // Conditionally serializable
private volatile V value; // 全部状态就这一个字段
public final V get() {
return value;
}
public final boolean compareAndSet(V expectedValue, V newValue) {
return VALUE.compareAndSet(this, expectedValue, newValue); // 比较的是引用 ==, 与 equals 无关
}
// ...
public final V getAndUpdate(UnaryOperator<V> updateFunction) {
V prev = get(), next = null;
for (boolean haveNext = false;;) { // 经典 CAS 自旋: 读旧值 -> 算新值 -> CAS, 失败重来
if (!haveNext)
next = updateFunction.apply(prev); // 函数可能被重复调用, 所以要求无副作用
if (weakCompareAndSetVolatile(prev, next))
return prev;
haveNext = (prev == (prev = get())); // 重读后引用没变就复用已算好的 next, 省一次函数调用
}
}
// ... set / getAndSet / compareAndExchange 等都是对 VALUE 的一行转发, 省略
}因为比较的只是引用, 它有 ABA 问题: 值从 A 改到 B 又改回 A, 期间发生过的事 CAS 察觉不到。对不可变对象的整体替换(每次 new 新对象)这通常无害; 在乎的话用 AtomicStampedReference, 给引用配一个 int 版本号一起 CAS。