当前位置：首页 > news >正文

并发编程——累加器

news 来源：原创 2025/6/8 12:21:41

1 AtomicLong

1.1 核心功能

1.2 实现原理：

(1）基于 Unsafe 的底层操作

(2) volatile字段的内存可见性

(3）CAS 操作与 ABA 问题

1.3 性能分析

1.4 使用场景

2 LongAdder

核心设计原理

1 分段存储

2 分散更新策略

3.处理高竞争

1 AtomicLong

AtomicLong 是一个基于 CAS操作的原子类，用于在多线程环境下对 long 类型变量进行无锁的原子性操作。它通过底层的 Unsafe 类实现高效的原子更新，解决传统 synchronized 关键字带来的性能开销问题。

1.1 核心功能

AtomicLong 支持以下原子操作：

方法	功能	实现原理
`add(long delta)`	原子性地将变量增加 `delta`	CAS 操作
`incrementAndGet()`	原子性自增 1 并返回新值	`add(1L)`
`decrementAndGet()`	原子性自减 1 并返回新值	`add(-1L)`
`getAndAdd(long delta)`	原子性增加 `delta` 并返回原值	CAS 操作
`getAndSet(long newValue)`	原子性设置新值并返回原值	CAS 操作
`compareAndSet(long expect, long update)`	原子性将值从 `expect` 更新为 `update` （如果相等）	CAS 操作
`get()`	获取当前值（非原子，但保证可见性）	直接读取 `value` 字段

1.2 实现原理：

(1）基于 `Unsafe` 的底层操作

AtomicLong 的所有原子方法均通过 sun.misc.Unsafe 类的底层原生指令实现，例如：

objectFieldOffset：获取long value字段在对象内存中的偏移量

getAndAddLong：通过CPU的CAS指令（如 cmpxchgq）原子性更新内存中的值。

(2) volatile字段的内存可见性

AtomicLong 的 value 字段声明为 volatile ，确保一个线程的修改对其它线程可见：

(3）CAS 操作与 ABA 问题

CAS 操作

CAS（Compare-And-Swap）包含三个步骤：

比较：读取内存中的旧值 expect。
交换：如果旧值等于 expect，则将新值 update 写入内存；否则不操作。
返回结果：返回内存中的旧值。

ABA 问题
场景：变量从 A → B → A，此时 compareAndSet(A, C) 会错误地认为值未变，导致更新失败。
解决方案：

AtomicStampedReference：引入版本戳（stamp）标记变量的修改次数，解决 ABA 问题。
LongAdder：通过分散更新压力避免单一变量的频繁 CAS 冲突。

1.3 性能分析

优势

无锁操作：避免线程阻塞和上下文切换，性能优于 synchronized。
读写高效：get() 方法是非原子的，但保证内存可见性，读取速度极快。
适用场景：低到中等并发场景，如计数器、单变量累加器。

局限性

高并发瓶颈：当多个线程激烈竞争同一变量时，CAS 失败次数剧增，导致自旋开销（spin-wait）。
ABA 问题：需额外处理或改用 AtomicStampedReference。

1.4 使用场景

低竞争环境：如单机多线程的计数器、序列号生成。

需要强原子性保证：如银行账户扣款、分布式锁的版本控制。

2 LongAdder

设计目标：LongAdder是Java 8引入的原子类，属于 java.util.concurrent.atomic 包，专为高并发场景下的累加操作优化。它的核心目标是解决AtomicLong 在及极高并发下的性能瓶颈——通过分散压力来减少线程间的CAS冲突。

核心设计原理

1 分段存储

cells数组: LongAdder维护一个动态增长的long[]cells数组，每个元素成为一个cell，用于存储部分累加值。
base变量：所有cell之外的累加值存储在base中，默认情况下，单个线程的增量优先base，当base发生竞争时，才会分配新的cell。

2 分散更新策略

优先更新 base（①）：
1. 若 cells 未初始化（null），或通过 casBase 成功将 base 加上 x，直接返回。casBase 是 Striped64 提供的原子操作，用于更新 base。
处理 cells 分段（②-④）：
1. 哈希索引计算：getProbe() 返回与当前线程绑定的哈希值（通过 ThreadLocalRandom 生成），并与数组长度掩码按位与，得到目标 cell 的索引。
2. CAS 更新 cell：若目标 cell 为空，通过 casCell 初始化为新值；否则尝试原子性增加 cell 的值。
3. 竞争标记：若 CAS 失败（uncontended = false），表示存在线程竞争。

3.处理高竞争

调用 longAccumulate 方法，可能触发以下操作：

动态扩容：若 cells 数组过小，倍增其大小并将 base 值分布到新 cell 中。
批量写入：将当前线程的增量暂存到临时变量，直到找到可写入的 cell。

public void add(long x) {
    Cell[] as; long b, v; int m; Cell a;
    if ((as = cells) != null || !casBase(b = base, b + x)) { // ① 优先尝试更新 base
        boolean uncontended = true;
        if (as == null || (m = as.length - 1) < 0 || // ② 处理未初始化或空数组
                (a = as[getProbe() & m]) == null ||     // ③ 计算哈希索引并获取目标 cell
                !(uncontended = a.cas(v = a.value, v + x))) { // ④ CAS 更新 cell
            longAccumulate(x, null, uncontended); // ⑤ 处理竞争，触发动态扩容
        }
    }
}

sum方法：返回base值和cells数组的总和

public long sum() {
    Cell[] as = cells; Cell a;
    long sum = base;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

reset方法：重置变量，使总和保持为零。

此方法可能是创建新 adder 的有用替代方法，但仅在没有并发更新时有效。由于此方法本质上是 racy，因此仅当已知没有线程同时更新时，才应使用它。

public void reset() {
    Cell[] as = cells; Cell a;
    base = 0L;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                a.value = 0L;
        }
    }
}

sumThenReset方法：返回当前总和后清零，适用于离线统计场景

实际上等效于 sum 后跟 reset。例如，在多线程计算之间的 static points 期间，此方法可能适用。如果存在与此方法并发的更新，则不能保证返回的值是重置之前出现的最终值。

public long sumThenReset() {
    Cell[] as = cells; Cell a;
    long sum = base;
    base = 0L;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null) {
                sum += a.value;
                a.value = 0L;
            }
        }
    }
    return sum;
}