无锁编程与LongAdder：高并发计数器的性能优化

// 1000个线程，每个线程累加100万次
AtomicLong counter = new AtomicLong(0);

ExecutorService executor = Executors.newFixedThreadPool(1000);
for (int i = 0; i < 1000; i++) {
    executor.submit(() -> {
        for (int j = 0; j < 1_000_000; j++) {
            counter.incrementAndGet();  // CAS操作
        }
    });
}

// 耗时：约 15秒（高并发场景）

// 同样的场景，使用LongAdder
LongAdder counter = new LongAdder();

ExecutorService executor = Executors.newFixedThreadPool(1000);
for (int i = 0; i < 1000; i++) {
    executor.submit(() -> {
        for (int j = 0; j < 1_000_000; j++) {
            counter.increment();  // 分段累加
        }
    });
}

long result = counter.sum();  // 最终求和
// 耗时：约 2秒（高并发场景）

// 性能提升：约 7.5倍！

// AtomicLong：所有线程竞争同一个变量
[AtomicLong] ← 线程1
             ← 线程2
             ← 线程3
             ← 线程4
// 高并发：大量CAS失败，性能下降

// LongAdder：分段累加，减少竞争
[Cell0] ← 线程1
[Cell1] ← 线程2
[Cell2] ← 线程3
[Cell3] ← 线程4
// 最终求和：Cell0 + Cell1 + Cell2 + Cell3

// 简化版LongAdder实现
public class SimpleLongAdder {
    // 基础值（低并发时使用）
    private volatile long base;

    // Cell数组（高并发时使用）
    private volatile Cell[] cells;

    // Cell内部类：填充避免伪共享
    @sun.misc.Contended
    static final class Cell {
        volatile long value;

        Cell(long x) { value = x; }

        // CAS更新
        final boolean cas(long cmp, long val) {
            return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
        }
    }

    public void add(long x) {
        Cell[] cs;
        long b, v;
        int m;
        Cell c;

        // 1. 如果cells为空，尝试直接累加到base
        if ((cs = cells) == null) {
            if (casBase(b = base, b + x)) {
                return;  // 成功，直接返回
            }
        }

        // 2. cells不为空，或base累加失败（高并发）
        // 根据线程哈希值，选择一个Cell
        int h = ThreadLocalRandom.getProbe();
        c = cs[(h & m)];

        // 3. 尝试CAS累加到该Cell
        if (c.cas(v = c.value, v + x)) {
            return;  // 成功
        }

        // 4. 失败：说明竞争激烈，需要扩容
        longAccumulate(x);  // 扩容逻辑
    }

    public long sum() {
        Cell[] cs = cells;
        long sum = base;

        // 遍历所有Cell，求和
        if (cs != null) {
            for (Cell c : cs) {
                if (c != null) {
                    sum += c.value;
                }
            }
        }

        return sum;
    }
}

public class PerformanceTest {

    private static final int ITERATIONS = 10_000_000;

    public static void main(String[] args) throws Exception {
        // 测试不同线程数
        int[] threadCounts = {1, 2, 4, 8, 16, 32, 64};

        for (int threadCount : threadCounts) {
            System.out.println("线程数：" + threadCount);

            // AtomicLong测试
            long atomicTime = testAtomicLong(threadCount);
            System.out.println("  AtomicLong耗时：" + atomicTime + "ms");

            // LongAdder测试
            long adderTime = testLongAdder(threadCount);
            System.out.println("  LongAdder耗时：" + adderTime + "ms");

            System.out.println("  性能提升：" + (atomicTime / (double) adderTime) + "倍\n");
        }
    }

    private static long testAtomicLong(int threadCount) throws Exception {
        AtomicLong counter = new AtomicLong(0);
        return runTest(threadCount, () -> counter.incrementAndGet());
    }

    private static long testLongAdder(int threadCount) throws Exception {
        LongAdder counter = new LongAdder();
        return runTest(threadCount, () -> counter.increment());
    }

    private static long runTest(int threadCount, Runnable task) throws Exception {
        CountDownLatch latch = new CountDownLatch(threadCount);
        long start = System.currentTimeMillis();

        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                for (int j = 0; j < ITERATIONS / threadCount; j++) {
                    task.run();
                }
                latch.countDown();
            }).start();
        }

        latch.await();
        return System.currentTimeMillis() - start;
    }
}

LongAdder adder = new LongAdder();

// 1. 增加
adder.increment();              // +1
adder.decrement();              // -1
adder.add(10);                  // +10

// 2. 获取结果
long sum = adder.sum();         // 求和（非原子操作）
long sumThenReset = adder.sumThenReset();  // 求和并重置

// 3. 重置
adder.reset();                  // 重置为0

// 4. 转换
long value = adder.longValue(); // 转为long（等同于sum）
int intValue = adder.intValue(); // 转为int

// ❌ 错误用法：sum()不是原子操作
LongAdder adder = new LongAdder();
adder.add(10);

if (adder.sum() < 100) {  // 非原子
    adder.add(50);        // 可能超过100
}

// ✅ 正确用法：LongAdder适用于统计场景
// 场景1：QPS统计
LongAdder requestCount = new LongAdder();
public void handleRequest() {
    requestCount.increment();  // 每次请求+1
}

public long getQPS() {
    long count = requestCount.sumThenReset();  // 获取并重置
    return count;  // 每秒请求数
}

// 场景2：并发计数
LongAdder errorCount = new LongAdder();
public void logError() {
    errorCount.increment();
}

// ❌ 不适合：需要精确原子操作的场景
// 例如：银行账户余额、库存扣减

// AtomicLong：需要精确原子操作
// 1. 生成唯一ID
AtomicLong idGenerator = new AtomicLong(0);
long id = idGenerator.incrementAndGet();  // 必须原子

// 2. 库存扣减
AtomicLong stock = new AtomicLong(100);
if (stock.decrementAndGet() >= 0) {
    // 扣减成功
}

// LongAdder：统计场景
// 1. QPS/TPS统计
LongAdder requestCount = new LongAdder();
requestCount.increment();  // 高并发

// 2. 错误计数
LongAdder errorCount = new LongAdder();
errorCount.increment();

// 3. 访问量统计
LongAdder pageViewCount = new LongAdder();
pageViewCount.increment();

// LongAdder只能做加法
LongAdder adder = new LongAdder();
adder.add(10);

// LongAccumulator支持自定义累加函数
LongAccumulator accumulator = new LongAccumulator((x, y) -> x + y, 0);
accumulator.accumulate(10);  // 累加

// 求最大值
LongAccumulator max = new LongAccumulator(Long::max, Long.MIN_VALUE);
max.accumulate(100);
max.accumulate(200);
max.accumulate(50);
System.out.println(max.get());  // 200

// 求最小值
LongAccumulator min = new LongAccumulator(Long::min, Long.MAX_VALUE);
min.accumulate(100);
min.accumulate(50);
min.accumulate(200);
System.out.println(min.get());  // 50

public class QPSMonitor {
    // 统计当前秒的请求数
    private LongAdder currentSecondCount = new LongAdder();

    // 统计历史最大QPS
    private LongAccumulator maxQPS = new LongAccumulator(Long::max, 0);

    // 记录每秒请求数
    @Scheduled(fixedRate = 1000)  // 每秒执行
    public void recordQPS() {
        long qps = currentSecondCount.sumThenReset();
        maxQPS.accumulate(qps);  // 更新最大QPS

        System.out.println("当前QPS：" + qps);
        System.out.println("历史最大QPS：" + maxQPS.get());
    }

    // 处理请求
    public void handleRequest() {
        currentSecondCount.increment();
    }
}

// CPU缓存行通常是64字节
// 如果两个变量在同一缓存行，会互相干扰

class Counter {
    volatile long value1;  // 8字节
    volatile long value2;  // 8字节
    // value1和value2可能在同一缓存行（64字节）
}

// 线程1修改value1 → 缓存行失效 → 线程2的value2缓存也失效
// 导致性能下降（虽然两个变量逻辑上独立）

// LongAdder的Cell使用@Contended避免伪共享
@sun.misc.Contended
static final class Cell {
    volatile long value;
}

// @Contended会在value前后填充128字节（缓存行对齐）
// 确保每个Cell独占缓存行，避免伪共享

// 需要开启JVM参数：-XX:-RestrictContended