并发设计模式实战系列(14)：CAS（无锁编程）

┌───────────────┐      ┌───────────────┐      ┌───────────────┐
│  内存值 (V)    │───>  │  预期值 (A)    │───>  │  新值 (B)      │
└───────────────┘      └───────────────┘      └───────────────┘
      │                      │                      │
      └───── 比较 V == A ─────┘                      │
            │                                      │
            ▼                                      ▼
       true: 更新V=B                        false: 操作失败

import java.util.concurrent.atomic.AtomicInteger;
public class CASCounter {
    private final AtomicInteger value = new AtomicInteger(0);
    // 无锁递增
    public int increment() {
        int oldValue, newValue;
        do {
            oldValue = value.get();      // 读取当前值（A）
            newValue = oldValue + 1;    // 计算新值（B）
        } while (!value.compareAndSet(oldValue, newValue)); // CAS操作
        return newValue;
    }
    public static void main(String[] args) {
        CASCounter counter = new CASCounter();
        
        // 模拟10个线程并发计数
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    counter.increment();
                }
            }).start();
        }
        
        // 等待所有线程完成
        try { Thread.sleep(2000); } catch (InterruptedException e) {}
        System.out.println("Final value: " + counter.value.get()); // 正确输出10000
    }
}

import sun.misc.Unsafe;
import java.lang.reflect.Field;
class ManualCAS {
    private volatile int value;
    private static final Unsafe unsafe;
    private static final long valueOffset;
    static {
        try {
            Field f = Unsafe.class.getDeclaredField("theUnsafe");
            f.setAccessible(true);
            unsafe = (Unsafe) f.get(null);
            valueOffset = unsafe.objectFieldOffset(ManualCAS.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }
    public boolean compareAndSwap(int expected, int newValue) {
        return unsafe.compareAndSwapInt(this, valueOffset, expected, newValue);
    }
}

// 使用ThreadLocal分散热点
ThreadLocalRandom.current().nextInt(10); // 随机退避时间
// 伪代码示例
while (!casUpdate()) {
    int backoff = ThreadLocalRandom.current().nextInt(10);
    LockSupport.parkNanos(backoff); // 随机休眠
}

AtomicStampedReference<String> ref = new AtomicStampedReference<>("A", 0);
// 更新时检查版本号
int stamp = ref.getStamp();
String current = ref.getReference();
ref.compareAndSet(current, "B", stamp, stamp + 1);

// CAS成功率统计
AtomicLong successCount = new AtomicLong();
AtomicLong totalAttempts = new AtomicLong();
public int incrementWithStats() {
    int oldValue, newValue;
    do {
        totalAttempts.incrementAndGet();
        oldValue = value.get();
        newValue = oldValue + 1;
    } while (!value.compareAndSet(oldValue, newValue));
    successCount.incrementAndGet();
    return newValue;
}

// 使用LongAdder替代AtomicLong（JDK8+）
LongAdder hitCounter = new LongAdder();
void recordHit() {
    hitCounter.increment(); // 内部使用分段CAS优化
}

// 基于AtomicReference的MPSC队列（多生产者单消费者）
class LockFreeQueue<E> {
    private static class Node<E> {
        E item;
        AtomicReference<Node<E>> next;
        Node(E item) { this.item = item; }
    }
    private final AtomicReference<Node<E>> head;
    private final AtomicReference<Node<E>> tail;
    public void enqueue(E item) {
        Node<E> newNode = new Node<>(item);
        Node<E> oldTail;
        do {
            oldTail = tail.get();
        } while (!tail.compareAndSet(oldTail, newNode));
        oldTail.next.set(newNode);
    }
}

// 使用AtomicInteger实现线程安全状态机
enum State { INIT, RUNNING, STOPPED }
AtomicInteger currentState = new AtomicInteger(State.INIT.ordinal());
boolean start() {
    return currentState.compareAndSet(
        State.INIT.ordinal(), 
        State.RUNNING.ordinal()
    );
}

// 错误示例：线程间互相导致CAS失败
void transfer(AtomicInteger acc1, AtomicInteger acc2, int amount) {
    while (true) {
        int old1 = acc1.get();
        if (!acc1.compareAndSet(old1, old1 - amount)) continue;
        
        int old2 = acc2.get();
        if (!acc2.compareAndSet(old2, old2 + amount)) {
            acc1.compareAndSet(old1 - amount, old1); // 回滚导致活锁
            continue;
        }
        break;
    }
}
// 解决方案：引入随机退避
if (!acc2.compareAndSet(old2, old2 + amount)) {
    acc1.compareAndSet(old1 - amount, old1);
    Thread.sleep(ThreadLocalRandom.current().nextInt(10)); // 关键点
    continue;
}

// 使用@Contended注解（JDK8+）
import jdk.internal.vm.annotation.Contended;
class CounterCell {
    @Contended  // 避免伪共享
    volatile long value;
}
// 或手动填充（兼容旧版本）
class ManualPadding {
    volatile long value;
    long p1, p2, p3, p4, p5, p6; // 填充至64字节（典型缓存行大小）
}

是否共享变量修改？
  ├─否 → volatile足够
  └─是 → 是否满足以下全部条件？
          ├─是 → 使用CAS
          │   ├─线程竞争适中（<20线程）
          │   ├─操作足够简单（单变量修改）
          │   └─能容忍偶尔失败
          └─否 → 选择锁
              ├─需要条件等待 → ReentrantLock
              └─简单同步 → synchronized

// 手动插入内存屏障示例（Unsafe类）
public class MemoryBarrierDemo {
    private volatile int flag;
    private int data;
    
    void publish() {
        data = 42;                          // 普通写
        Unsafe.getUnsafe().storeFence();    // 写屏障
        flag = 1;                           // volatile写
    }
    
    void consume() {
        while (flag != 1) { /* 自旋 */ }    // volatile读
        Unsafe.getUnsafe().loadFence();     // 读屏障
        System.out.println(data);           // 保证看到42
    }
}

// 通过局部变量避免CAS竞争
public class LocalCounter {
    public long sum() {
        final long[] localSum = new long[1]; // 栈上分配（逃逸分析优化）
        
        IntStream.range(0, 100).parallel().forEach(i -> {
            localSum[0] += i; // 线程本地计算
        });
        
        return localSum[0]; // 最终合并
    }
}

// 热点代码的循环展开（JIT自动优化）
@Benchmark
public void testCAS(Blackhole bh) {
    AtomicInteger counter = new AtomicInteger();
    for (int i = 0; i < 1000; i++) {
        counter.incrementAndGet(); // 可能被JIT展开为批量CAS
    }
    bh.consume(counter);
}

-XX:+UnlockDiagnosticVMOptions 
-XX:+PrintAssembly 
-XX:+LogCompilation

// 伪共享解决方案对比
@Contended // JDK8+官方方案（需开启-XX:-RestrictContended）
class ContendedCounter {
    volatile long value1;
    volatile long value2;
}
// 手动填充方案（兼容旧JDK）
class ManualPaddingCounter {
    volatile long value1;
    long p1, p2, p3, p4, p5, p6, p7; // 填充56字节
    volatile long value2;
}

cat /sys/devices/system/cpu/cpu0/cache/index0/coherency_line_size

// 绑定线程到指定CPU核（Linux）
public class NumaAffinity {
    static {
        System.loadLibrary("numa_affinity");
    }
    
    public native void bindToCore(int coreId);
    public static void main(String[] args) {
        new NumaAffinity().bindToCore(2); // 绑定到2号核
        // 执行敏感计算...
    }
}

-XX:+UseNUMA 
-XX:+UseParallelGC

// 利用CAS批量操作减少总线锁定
class BatchUpdater {
    private final AtomicLongArray values;
    
    void batchUpdate(int[] indices, long[] updates) {
        for (int i = 0; i < indices.length; i++) {
            int idx = indices[i];
            long expect, update;
            do {
                expect = values.get(idx);
                update = expect + updates[i];
            } while (!values.compareAndSet(idx, expect, update));
        }
    }
}

// 终极优化版计数器（结合所有优化技术）
@Contended
public class UltimateCounter {
    @Contended
    private volatile long cell1;
    @Contended
    private volatile long cell2;
    private static final sun.misc.Unsafe UNSAFE;
    private static final long cell1Offset;
    
    static {
        try {
            Field f = sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
            f.setAccessible(true);
            UNSAFE = (sun.misc.Unsafe) f.get(null);
            cell1Offset = UNSAFE.objectFieldOffset(
                UltimateCounter.class.getDeclaredField("cell1"));
        } catch (Exception e) { throw new Error(e); }
    }
    public void increment() {
        long expect;
        do {
            expect = UNSAFE.getLongVolatile(this, cell1Offset);
        } while (!UNSAFE.compareAndSwapLong(
            this, cell1Offset, expect, expect + 1));
    }
}