【HUC】原子类 LongAdder：从 add() 到 longAccumulate() 源码解析

LongAdder的设计哲学是用空间换时间通过分散热点来应对高并发。add()方法作为入口其逻辑清晰地体现了这一思想尽可能在无竞争或低竞争路径上快速返回将复杂的竞争处理“委托”给longAccumulate()。add(long x)快速路径与竞争委托LongAdder本身没有定义add它继承自Striped64。add方法的代码不长但每一行都对应着一种并发场景。publicvoidadd(longx){Cell[]as;longb,v;intm;Cella;// 条件一cells 已初始化 (as ! null)// 条件二!casBase(b base, b x) 意味着对 base 的 CAS 更新失败if((ascells)!null||!casBase(bbase,bx)){booleanuncontendedtrue;// 初始假定无竞争// 进入此分支意味着要么 cells 已存在要么 base 的 CAS 失败了// 条件一cells 为 nullbase CAS 失败但 cells 尚未初始化// 条件二cells 长度为 0理论上不应发生// 条件三当前线程 hash 命中的 Cell 为 null需要新建 Cell// 条件四!a.cas(v a.value, v x) 对命中的 Cell 进行 CAS 更新失败if(asnull||(nas.length)1||(aas[(n-1)(mgetProbe())])null||!(uncontendeda.cas(va.value,vx)))longAccumulate(x,null,uncontended);}}执行流程解读低竞争快路径如果cells尚未初始化as null方法会首先尝试对全局的base进行 CAS 累加casBase。在并发度极低时这一步就能成功add方法直接返回没有任何额外开销。base CAS 失败 → 触发复杂逻辑如果cells已经初始化或者base的 CAS 操作失败说明有竞争就会进入if块准备在cells数组上操作。Cell 层面的快速路径在进入longAccumulate之前add会做最后一次“挣扎”。它通过getProbe()获取当前线程的探测哈希值m定位到cells数组中对应的槽位as[(n-1) m]。如果该槽位已经存在一个Cell对象它会尝试直接对这个Cell的value进行 CAS 累加。这次 CAS 的结果会被记录在uncontended变量中。如果这次 CAS 成功方法同样直接返回无需进入更重量级的longAccumulate。委托给longAccumulate只有当出现以下任一情况时才会调用longAccumulatecells数组尚未初始化但baseCAS 已失败。线程命中的槽位为null需要创建新的Cell。对已存在的Cell进行 CAS 更新失败uncontended false表明该Cell也出现了竞争。关键点add方法通过casBase和Cell.cas两次尝试尽可能避免进入longAccumulate。传入的wasUncontended参数即uncontended变量告知了longAccumulate“在你接手之前我已经在 Cell 上尝试过 CAS 并且失败了”这对于longAccumulate的后续决策至关重要。longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended)竞争处理核心这个方法由Striped64提供是处理初始化、创建Cell、扩容以及解决竞争的总控逻辑。在LongAdder的调用中fn始终为null代表简单的加法但该方法也服务于支持自定义函数的LongAccumulator。finalvoidlongAccumulate(longx,LongBinaryOperatorfn,booleanwasUncontended){inth;// 1. 确保当前线程拥有一个非零的 probe 哈希值if((hgetProbe())0){ThreadLocalRandom.current();// 强制初始化线程的 probehgetProbe();wasUncontendedtrue;// 新分配 probe视为无竞争}booleancollidefalse;// 扩容意向标志for(;;){// 2. 无限自旋直至成功Cell[]as;Cella;intn;longv;// CASE 1: cells 数组已初始化if((ascells)!null(nas.length)0){// ...}// CASE 2: cells 未初始化尝试初始化elseif(cellsBusy0cellsascasCellsBusy()){// ...}// CASE 3: 回退直接累加到 baseelseif(casBase(vbase,((fnnull)?vx:fn.applyAsLong(v,x))))break;}}longAccumulate的主体是一个for (;;)自旋循环它的内部被划分为三个主要的CASE分别对应cells的不同状态。前置操作线程 Probe 的初始化getProbe()返回当前线程的探测哈希值probe。如果为0意味着该线程尚未被分配probe。此时会调用ThreadLocalRandom.current()强制初始化并重新获取probe。同时wasUncontended被强制设为true。原因是一个新分配probe的线程其哈希值默认会映射到cells[0]这不被视为一次真正的竞争失败因此重置竞争标志给它一次“干净”的重试机会。CASE 1cells已初始化 —— 创建Cell、重试或扩容这是最复杂的路径当cells数组存在且长度大于0时进入。if((ascells)!null(nas.length)0){// CASE 1.1: 命中的槽位为 null需要创建新 Cellif((aas[(n-1)h])null){if(cellsBusy0){// 无锁状态CellrnewCell(x);// 乐观地预先创建if(cellsBusy0casCellsBusy()){// 获取锁booleancreatedfalse;try{Cell[]rs;intm,j;// 双重检查防止在获取锁期间其他线程已在该位置创建了 Cellif((rscells)!null(mrs.length)0rs[j(m-1)h]null){rs[j]r;createdtrue;}}finally{cellsBusy0;// 释放锁}if(created)break;// 创建成功退出自旋continue;// 槽位已被其他线程占用重试}}collidefalse;// 命中空槽位不需要扩容}// CASE 1.2: 之前已知 CAS 失败重置标志后重试elseif(!wasUncontended)wasUncontendedtrue;// 继续循环使用重新哈希后的 probe 重试// CASE 1.3: 对命中的 Cell 进行 CAS 更新elseif(a.cas(va.value,((fnnull)?vx:fn.applyAsLong(v,x))))break;// 更新成功退出自旋// CASE 1.4: 已达到扩容上限或 cells 已被其他线程修改放弃本次扩容意图elseif(nNCPU||cells!as)collidefalse;// CASE 1.5: 标记扩容意向但尚未真正扩容elseif(!collide)collidetrue;// CASE 1.6: 真正执行扩容elseif(cellsBusy0casCellsBusy()){try{if(cellsas){// 再次检查防止重复扩容Cell[]rsnewCell[n1];// 容量翻倍for(inti0;in;i)rs[i]as[i];// 迁移已有 Cellcellsrs;}}finally{cellsBusy0;}collidefalse;continue;// 扩容后重试}hadvanceProbe(h);// ★ 关键一步重新计算当前线程的 probe}CASE 1 的逻辑链条CASE 1.1空槽位当前线程的probe映射到一个null槽位。此时会在cellsBusy自旋锁的保护下通过双重检查rs[j] null安全地创建一个新的Cell并放入。cellsBusy是一个通过 CAS 维护的自旋锁用于保护cells数组的结构性修改初始化、扩容、创建 Cell。CASE 1.2重置竞争标志如果add方法传入的wasUncontended为false表示在add中 Cell CAS 已失败这里会将其重置为true。这相当于给该线程一次“申诉”机会之前失败是因为probe对应的 Cell 有竞争现在我会重新计算probe在最后一行advanceProbe(h)换一个 Cell 再试。CASE 1.3Cell CAS 成功对命中的Cell的value进行 CAS 累加。这是最常见的成功路径成功后直接break退出自旋。CASE 1.4 - 1.6扩容决策如果n NCPU数组长度已达 CPU 核数上限或cells引用已发生变化则放弃扩容collide false。如果collide为false将其设为true。这是一个扩容意向标记表示“我刚刚在这个 Cell 上又失败了可能真的需要扩容了”。如果collide已经是true意味着上次失败后重试仍然失败并且成功获取了cellsBusy锁则真正执行扩容将数组长度翻倍n 1并迁移所有已有的Cell引用。h advanceProbe(h)无论本轮CASE 1的哪个分支被执行除了成功break或continue到扩容外最后都会执行这行代码。它使用一个伪随机函数重新计算当前线程的probe。这是实现“分散热点”的核心机制——当在一个Cell上竞争失败后线程会“换一个地方”再试而不是死磕同一个槽位。CASE 2cells未初始化 —— 安全地完成初始化当cells为null或长度为0且cellsBusy锁可用时进入此分支。elseif(cellsBusy0cellsascasCellsBusy()){booleaninitfalse;try{// 双重检查防止在获取锁期间其他线程已完成初始化if(cellsas){Cell[]rsnewCell[2];// 初始容量为 2rs[h1]newCell(x);// 将当前线程的增量放入对应的 Cellcellsrs;inittrue;}}finally{cellsBusy0;}if(init)break;// 初始化成功退出自旋}这是cells数组的懒初始化路径。只有在baseCAS 失败后add才会委托到这里。初始化过程同样受cellsBusy锁保护并包含cells as的双重检查以确保并发环境下初始化操作的原子性。初始数组长度为2当前线程的增量x会被直接放入rs[h 1]对应的Cell中。CASE 3回退 —— 直接更新base这是最后的保底路径。elseif(casBase(vbase,((fnnull)?vx:fn.applyAsLong(v,x))))break;当CASE 1cells存在但处理失败和CASE 2无法获取锁来初始化cells都走不通时意味着当前有另一个线程正在持有cellsBusy锁可能在初始化或扩容或者cells已被其他线程初始化。此时当前线程会选择回退到最基础的base上进行 CAS 累加。如果这个 CAS 成功就退出自旋如果失败则继续循环重试。 核心机制与设计思想总结从add到longAccumulate的整个流程贯穿了以下几个关键设计分层竞争处理性能敏感路径被极致优化。baseCAS →CellCAS →longAccumulate每一层都试图以最低成本解决问题将重量级操作限制在真正需要的时候。cellsBusy自旋锁一个轻量级的 CAS 锁仅用于保护cells数组的结构性修改初始化、扩容、创建Cell。对Cell内部value的更新并不需要此锁而是通过Cell自身的 CAS 完成。advanceProbe重哈希这是“分散”策略的引擎。在竞争失败或创建新Cell后线程的probe会变化从而在后续重试中可能命中cells数组中的不同槽位动态地分散访问压力。有界扩容 (n NCPU)cells数组的长度不会无限增长其上限是 CPU 核数。这避免了过度扩容带来的内存和缓存开销因为在核数已满的情况下更多的槽位也无法带来真正的并行度提升。Cell的缓存行填充Cell是AtomicLong的一个“填充变体”Padded variant。它的value字段前后有额外的填充字段目的是防止伪共享False Sharing。这确保了对不同Cell的更新不会因为落在同一个 CPU 缓存行上而相互干扰是高性能的关键底层优化。sum()的结果最终LongAdder.sum()返回的是base的值与cells数组中所有非null的Cell的value之和。需要注意sum()方法在并发更新时不保证原子性它返回的是一个“弱一致”的快照值。LongAdder通过这套精密的协作机制在高并发统计求和的场景下实现了远超AtomicLong的吞吐量其代价是在空间Cell数组、填充和读取时的数据一致性sum()非原子上做出了权衡。相关方法和变量定义流程图