承接的问题
8.2 看到了 LongAdder 的外部行为：base 加 cells、sum() 不是快照、每个 Cell 约 289 字节。本节打开引擎，回答三个实现问题：线程怎么知道自己该写哪个 Cell？cellsBusy 保护的临界区里到底做了什么，为什么不用阻塞锁？longAccumulate 在一个循环里如何区分"建槽、重试、扩容"三种情况？
这三个问题合起来就是高并发计数器的设计骨架：用线程本地的哈希把写分散，用一个只保护结构变化的短临界区兜底，用重试代替排队。
探针：线程到槽位的映射
LongAdder.add 里有一行 int index = getProbe()。这个 probe 不是 Striped64 自己的字段，它存在 java.lang.Thread 里，由 ThreadLocalRandom 维护：
// JDK 25 源码摘录，已删减
static final int getProbe() {
return U.getInt(JLA.currentCarrierThread(), PROBE);
}
static final int advanceProbe(int probe) {
probe ^= probe << 13; // xorshift
probe ^= probe >>> 17;
probe ^= probe << 5;
U.putInt(JLA.currentCarrierThread(), PROBE, probe);
return probe;
}
两个细节。第一，advanceProbe 是 xorshift：同一线程连续冲突时，低几位会变，从而映射到不同的槽。第二，读写的对象是 currentCarrierThread()，不是 Thread.currentThread()。源码在 localInit 里写得很清楚：“if virtual, share probe with carrier”——虚拟线程与它的载体线程共用同一个 probe。原因不难推：Cell[] 的槽位数按 CPU 核数封顶，映射的粒度本来就该是"谁在实际占用 CPU"，而不是"逻辑上有多少个任务"。
Striped64 与 ThreadLocalRandom 不在同一个包，前者怎么读到后者的包内字段？靠 SharedSecrets 桥：
// JDK 25 源码摘录，已删减
interface JavaUtilConcurrentTLRAccess {
int getThreadLocalRandomProbe();
int advanceThreadLocalRandomProbe(int r);
}
// JDK 25 源码摘录，已删减
private static final JavaUtilConcurrentTLRAccess TLR =
SharedSecrets.getJavaUtilConcurrentTLRAccess();
static final int getProbe() {
return TLR.getThreadLocalRandomProbe();
}
static final int advanceProbe(int probe) {
return TLR.advanceThreadLocalRandomProbe(probe);
}
为什么不把 Thread.threadLocalRandomProbe 改成 public？因为那是 java.lang.Thread 的内部布局，一旦公开就成了受兼容性约束的 API。SharedSecrets 是 JDK 内部的"好友通道"：接口在 jdk.internal.access，实现注册在嵌套类 ThreadLocalRandom$Access 的静态块里；getJavaUtilConcurrentTLRAccess() 发现字段为空时会 Class.forName 触发这个静态块，再返回注册好的实例。只有 java.base 内的可信代码拿得到。
本机用反射读出真实数值（--add-opens java.base/java.lang=ALL-UNNAMED，只用于观察）：
访问 TLR 之前 probe=0
ThreadLocalRandom.current() 之后 probe=-1640531527
调用 1000 次 nextInt 之后 probe=-1640531527（不变=true）
Striped64.advanceProbe(-1640531527) = 1359758873
调用后线程字段 probe=1359758873（已更新=true）
Striped64.getProbe() = 1359758873
-1640531527 就是 0x9E3779B9（黄金比例增量）的有符号表示，这是 PROBE_INCREMENT；调用 1000 次 nextInt 不会动 probe，说明 probe 和随机数生成是两条独立的流。probe 初值为 0，longAccumulate 的第一行 if (index == 0) { ThreadLocalRandom.current(); ... } 就是为它准备的：0 表示"还没初始化"，不能拿来索引。
cellsBusy：只在结构变化时用的自旋锁
Striped64 只有一把锁，而且它不是 ReentrantLock：
// JDK 25 源码摘录，已删减
final boolean casCellsBusy() {
return CELLSBUSY.compareAndSet(this, 0, 1);
}
它保护三件事：初始化 cells 数组、扩容数组、往空槽里安装新 Cell。三件事的共同点是"改结构"，不是"改值"。正常的累加走 Cell.cas，完全不碰这把锁。
为什么用自旋 CAS 而不是阻塞锁？因为临界区只有几条指令：分配一个 Cell、写进数组、把 cellsBusy 置回 0。让拿不到锁的线程 park 再 unpark，开销比这段临界区本身还大。所以源码的处理是：拿不到锁就不拿，去试 base、去试别的槽、或者 advanceProbe 换一个槽；等锁的人不会排队，也不会阻塞。
释放锁用的是普通写 cellsBusy = 0，不是 CAS。这是安全的，因为只有持锁线程会写它；cellsBusy 是 volatile，这次写本身带 release 语义，把临界区里对 cells 的修改发布出去。所有安装/扩容路径都在锁内做一次 cells == cs 或 rs[j] == null 的复检：等锁期间数组可能已经被别人换掉，不复检就会写进一个已经废弃的数组。
longAccumulate 的三条分支
longAccumulate 是整个 Striped64 的主循环。先看进入函数和"槽位为空"的处理：
// JDK 25 源码摘录，已删减
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended, int index) {
if (index == 0) {
ThreadLocalRandom.current(); // force initialization
index = getProbe();
wasUncontended = true;
}
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;
break;
}
} finally {
cellsBusy = 0;
}
continue; // Slot is now non-empty
}
}
collide = false;
}
槽位为空时，先"乐观地" new Cell(x)，再抢锁；抢到后复检槽位还是不是空的，是就把新 Cell 挂上去。new 放在锁外，是为了把可能触发 GC 的分配移出临界区；复检放在锁内，是因为等锁期间别人可能已经装好了。
接着是槽位非空时的 CAS 与扩容：
// JDK 25 源码摘录，已删减
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;
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);
} finally {
cellsBusy = 0;
}
collide = false;
continue; // Retry with expanded table
}
index = advanceProbe(index);
最后是表还没建时的初始化，以及退回 base：
// JDK 25 源码摘录，已删减
else if (cellsBusy == 0 && cells == cs && casCellsBusy()) {
try { // Initialize table
if (cells == cs) {
Cell[] rs = new Cell[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;
}
}
这条链上的几处控制流值得逐条拆：
wasUncontended是一个"免扩容"标记。调用方add已经试过一次Cell.cas并失败，才带着false进来；这里先把标记翻成true然后继续循环，只换探针、不扩容。一次失败可能只是瞬时碰撞，不值得为它翻倍数组。collide是"连续两次失败"的计数器。第一次 CAS 失败只把collide置true；同一个线程在同一轮里再失败一次，才进入扩容分支。扩容是 O(n) 的Arrays.copyOf，必须确认竞争是持续的。n >= NCPU || cells != cs时只重置collide，不扩容。槽位数封顶在 CPU 数，因为再多的槽也不会带来更多并行；cells != cs表示数组已被别人换掉，当前这一轮是基于旧数组的判断，直接作废。- 扩容只翻倍，不缩容。源码注释解释了不回收：长生命周期实例的竞争会复发，回收的槽还会被再次需要；短生命周期实例根本不值得回收。
- 每一轮末尾都
advanceProbe。同一槽位反复失败时，换一个哈希比原地重试更可能找到空闲槽。
一句话概括：先试一次，失败只换哈希；连续失败才扩容；到 CPU 数就停止扩容，改为在探针空间里找一个不冲突的映射。
手写一个 16 槽的 Striped64
把上面的骨架缩到最小：base + Cell[]、per-thread 探针、cellsBusy 自旋锁、翻倍扩容，最多 16 槽。为了让代码能放进一节，省略了 @Contended 填充和 doubleAccumulate：
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
/** 简化版 Striped64：最多 16 个 Cell，探针冲突后扩容。省略了 @Contended 填充。 */
public final class MiniStriped2 {
static final int MAX = 16;
static final class Cell {
volatile long value;
Cell(long v) { value = v; }
boolean cas(long c, long v) { return VALUE.compareAndSet(this, c, v); }
}
private volatile Cell[] cells;
private volatile long base;
private volatile int busy;
private int expansions;
private static final VarHandle CELLS, BASE, BUSY, VALUE;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
CELLS = l.findVarHandle(MiniStriped2.class, "cells", Cell[].class);
BASE = l.findVarHandle(MiniStriped2.class, "base", long.class);
BUSY = l.findVarHandle(MiniStriped2.class, "busy", int.class);
VALUE = l.findVarHandle(Cell.class, "value", long.class);
} catch (ReflectiveOperationException e) { throw new ExceptionInInitializerError(e); }
}
private static final ThreadLocal<Integer> PROBE = ThreadLocal.withInitial(() -> {
int p = ThreadLocalRandom.current().nextInt();
return p == 0 ? 1 : p;
});
static int advance(int p) { p ^= p << 13; p ^= p >>> 17; p ^= p << 5; PROBE.set(p); return p; }
public void add(long x) {
Cell[] cs; long b, v; int m; Cell c;
if ((cs = cells) != null || !BASE.compareAndSet(this, b = base, b + x)) {
int index = PROBE.get(); boolean uncontended = true;
if (cs == null || (m = cs.length - 1) < 0 || (c = cs[index & m]) == null ||
!(uncontended = c.cas(v = c.value, v + x)))
accumulate(x, uncontended, index);
}
}
private void accumulate(long x, boolean wasUncontended, int index) {
for (boolean collide = false;;) {
Cell[] cs; Cell c; int n; long v;
if ((cs = cells) != null && (n = cs.length) > 0) {
if ((c = cs[(n - 1) & index]) == null) {
if (busy == 0 && BUSY.compareAndSet(this, 0, 1)) {
try {
Cell[] rs; int m, j;
if ((rs = cells) != null && (m = rs.length) > 0 &&
rs[j = (m - 1) & index] == null) { rs[j] = new Cell(x); return; }
} finally { busy = 0; }
}
collide = false;
} else if (!wasUncontended) {
wasUncontended = true;
} else if (c.cas(v = c.value, v + x)) {
return;
} else if (n >= MAX || cells != cs) {
collide = false;
} else if (!collide) {
collide = true;
} else if (busy == 0 && BUSY.compareAndSet(this, 0, 1)) {
try { if (cells == cs) { CELLS.set(this, Arrays.copyOf(cs, n << 1)); expansions++; } }
finally { busy = 0; }
collide = false; continue;
}
index = advance(index);
} else if (busy == 0 && cells == cs && BUSY.compareAndSet(this, 0, 1)) {
try {
if (cells == cs) {
Cell[] rs = new Cell[2]; rs[index & 1] = new Cell(x);
CELLS.set(this, rs); expansions++; return;
}
} finally { busy = 0; }
} else if (BASE.compareAndSet(this, v = base, v + x)) {
return;
}
}
}
public long sum() {
Cell[] cs = cells; long s = base;
if (cs != null) for (Cell c : cs) if (c != null) s += c.value;
return s;
}
public static void main(String[] args) throws Exception {
int threads = Integer.parseInt(args[0]), ops = 1_000_000;
MiniStriped2 m = new MiniStriped2();
CountDownLatch start = new CountDownLatch(1);
Thread[] ts = new Thread[threads];
for (int t = 0; t < threads; t++)
ts[t] = Thread.ofPlatform().start(() -> {
try { start.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
for (int i = 0; i < ops; i++) m.add(1);
});
long t0 = System.nanoTime(); start.countDown();
for (Thread th : ts) th.join();
long expected = (long) threads * ops;
System.out.println("threads=" + threads + " sum=" + m.sum() + " 期望=" + expected
+ " cells=" + (m.cells == null ? 0 : m.cells.length)
+ " expansions=" + m.expansions + " 耗时=" + (System.nanoTime() - t0) / 1_000_000 + "ms");
}
}
本机运行（load average 68.96，10 核）：
threads=8 sum=8000000 期望=8000000 cells=8 expansions=3 耗时=79ms
threads=16 sum=16000000 期望=16000000 cells=8 expansions=3 耗时=115ms
两次都是 3 次扩容（2→4→8），总和精确。重复运行时 cells 还会出现 4 和 16：扩容取决于实际碰撞，不取决于线程数。collide 要求同一个线程在同一个槽上连续失败两次才翻倍；线程被调度得越散，失败一次后换个探针就成功了，表就停在小尺寸。16 线程只扩到 8 槽并不奇怪，两个线程共用一个槽也仍然是正确的，只是多几次 CAS 重试。耗时受本机负载影响很大，这里不作吞吐结论——唯一必须在所有运行里成立的是 sum() == 期望。
在 libraryd-conc 里落地
分馆借阅统计里有两个典型指标：累计借出量（纯加法）和单次借阅批次的最大值（非加法聚合）。前者用 LongAdder，后者用 LongAccumulator，两者共用 Striped64：
package com.plume.conc.core;
import java.util.concurrent.atomic.LongAccumulator;
import java.util.concurrent.atomic.LongAdder;
/** 单个分馆的借阅统计。borrows 是累加，maxBatch 是求最大值。 */
public final class BranchStats {
private final LongAdder borrows = new LongAdder();
private final LongAccumulator maxBatch = new LongAccumulator(Math::max, 0L);
/** batch 为本次借出的册数，必须为正。 */
public void record(int batch) {
if (batch <= 0) throw new IllegalArgumentException("batch must be positive");
borrows.add(batch);
maxBatch.accumulate(batch);
}
public long borrows() { return borrows.sum(); }
public long maxBatch() { return maxBatch.get(); }
}
LongAccumulator 的构造参数是 (LongBinaryOperator, identity)，Math::max 满足结合律和交换律，所以更新顺序不影响结果；0L 是单位元。如果换成减法或字符串拼接，结果就依赖执行顺序，不能用在 Striped64 上。maxBatch 只增不减；要统计"每分钟最大批次"，正确做法是每个窗口换一个新累加器，而不是试图清零——reset 在并发下不安全，8.2 已经说明。
库存快照版本号则不能用这两个类。它要求"读到的每个版本都是一个确定的历史状态"，只能用 AtomicLong 的 incrementAndGet/get。LongAdder.sum() 连"某一时刻的值"都不保证，拿它当版本号会让读者看到重复或跳变的版本。
设计的代价与边界
第一，不要在生产代码里手写 Striped64。本节这段一百来行的版本省掉了 @Contended、doubleAccumulate、探针与 ThreadLocalRandom 的共享、以及各种复检；任何一处省错，得到的就是"大多数时候对"的计数器。JDK 的实现有 JCStress 级别的验证，直接用 LongAdder/LongAccumulator。
第二，LongAccumulator 的函数必须满足结合律。源码不保证更新顺序，也不保证函数只被调用一次（CAS 失败会重算）。带副作用的函数会在重试时被执行多次。
第三，槽位不回收。cells 只扩不缩，一次高并发峰值会永久抬高内存占用；每个槽约 289 字节。
第四，虚拟线程共享载体探针。JDK 25 的 probe 挂在 carrier thread 上，大量虚拟线程跑在少量载体上时，它们会共用同一批槽位，分片效果退化。CPU 密集的计数本来就不该放在虚拟线程里；如果确实要，接受碰撞率上升，或者改用每个任务自己的局部计数再汇总。
第五，它不能替代序列号。Striped64 没有"读旧值并加一"的原子操作，sum() 也不保证单调到可以用来排序；sum() 也不是快照，这条在 8.2 已用实测数据说明。
小结
Striped64 用三样东西把单点竞争摊开：线程探针（ThreadLocalRandom 维护、虚拟线程与载体共享、经 SharedSecrets 桥接）决定写哪个槽；cellsBusy 一把只在建表、扩容、装槽时使用的自旋锁保护结构变化；longAccumulate 用一个 collide 状态机区分"换探针重试"和"翻倍扩容"，并在 NCPU 处封顶。手写简化版跑出的 8 槽/16 槽结果与总和都精确，但真正的工程结论是：理解它，然后使用它，不要重造它。
下一章从原子类转向 VarHandle：8.1 说 AtomicInteger 因为启动依赖没能用上它，9.1 就来看它到底提供了哪些访问模式。
阅读导航：上一节 8.2 LongAdder 的 Cell 分片 ｜ 下一节 9.1 VarHandle 的访问模式
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。