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

无锁化编程（Lock-Free Programming）分析

news 2025/9/18 12:56:27

无锁化编程（Lock-Free Programming）分析

概述

无锁化编程（Lock-Free Programming）是一种并发编程技术，它不使用传统的互斥锁（mutex）、信号量（semaphore）等阻塞同步原语来保证线程安全，而是通过原子操作、内存屏障和巧妙的算法设计来实现线程间的同步和协调。

定义

无锁化编程具有以下特征：

非阻塞性：线程不会因为等待锁而被阻塞
系统级进展保证：至少有一个线程能够在有限时间内完成操作
避免死锁：由于不使用锁，天然避免了死锁问题
高并发性能：在高竞争环境下通常表现更好

传统锁机制的问题

1. 性能问题

// 传统锁的性能瓶颈示例
public class CounterWithLock {private int count = 0;private final Object lock = new Object();public void increment() {synchronized(lock) {  // 获取锁可能导致线程阻塞count++;          // 临界区操作}                     // 释放锁}public int getCount() {synchronized(lock) {return count;}}
}

问题分析：

上下文切换开销：线程阻塞和唤醒需要系统调用
锁竞争：多个线程竞争同一把锁时性能急剧下降
优先级反转：高优先级线程可能被低优先级线程阻塞

2. 可靠性问题

死锁：多个锁的获取顺序不当可能导致死锁
活锁：线程不断重试但无法进展
饥饿：某些线程可能长时间无法获得锁

3. 可扩展性问题

串行化瓶颈：锁将并行操作串行化
缓存一致性开销：锁的获取和释放需要缓存同步

无锁化编程核心概念

1. 原子性（Atomicity）

原子操作是不可分割的操作，要么完全执行，要么完全不执行。

// Java中的原子操作示例
AtomicInteger atomicCounter = new AtomicInteger(0);// 原子递增操作
int newValue = atomicCounter.incrementAndGet();// 原子比较并交换操作
int expectedValue = 10;
int newValue = 20;
boolean success = atomicCounter.compareAndSet(expectedValue, newValue);

2. 可见性（Visibility）

一个线程对共享变量的修改，其他线程能够立即看到。

// 使用volatile保证可见性
public class VisibilityExample {private volatile boolean flag = false;public void writer() {flag = true;  // 写操作对其他线程立即可见}public void reader() {if (flag) {   // 能够看到writer线程的修改// 执行相应逻辑}}
}

3. 有序性（Ordering）

程序执行的顺序按照代码的先后顺序执行，防止指令重排序。

// 内存屏障示例
public class OrderingExample {private volatile int x = 0;private volatile int y = 0;public void method1() {x = 1;        // 操作1y = 1;        // 操作2 - volatile写入作为内存屏障}public void method2() {if (y == 1) { // 如果看到y=1assert x == 1; // 那么一定能看到x=1}}
}

原子操作与CAS

1. Compare-And-Swap (CAS) 原理

CAS是无锁编程的核心操作，它包含三个参数：

内存位置（V）：要更新的变量
预期值（A）：期望的当前值
新值（B）：要设置的新值

// CAS操作的伪代码
boolean compareAndSwap(int* ptr, int expected, int new_value) {if (*ptr == expected) {*ptr = new_value;return true;}return false;
}

2. CAS的硬件支持

现代CPU提供了硬件级别的CAS指令：

x86：CMPXCHG 指令
ARM：LDREX/STREX 指令对
SPARC：CAS 指令

3. Java中的CAS实现

import java.util.concurrent.atomic.AtomicInteger;public class CASExample {private AtomicInteger value = new AtomicInteger(0);// 无锁递增实现public int lockFreeIncrement() {int current, next;do {current = value.get();        // 读取当前值next = current + 1;           // 计算新值} while (!value.compareAndSet(current, next)); // CAS操作return next;}// 无锁加法实现public int lockFreeAdd(int delta) {int current, next;do {current = value.get();next = current + delta;} while (!value.compareAndSet(current, next));return next;}
}

4. CAS操作的性能特点

// 性能测试示例
public class CASPerformanceTest {private static final int THREAD_COUNT = 8;private static final int OPERATIONS_PER_THREAD = 1000000;// 使用CAS的计数器private AtomicInteger casCounter = new AtomicInteger(0);// 使用锁的计数器private int lockCounter = 0;private final Object lock = new Object();public void testCASPerformance() throws InterruptedException {// CAS性能测试long startTime = System.nanoTime();Thread[] casThreads = new Thread[THREAD_COUNT];for (int i = 0; i < THREAD_COUNT; i++) {casThreads[i] = new Thread(() -> {for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {casCounter.incrementAndGet();}});casThreads[i].start();}for (Thread thread : casThreads) {thread.join();}long casTime = System.nanoTime() - startTime;System.out.println("CAS Time: " + casTime / 1_000_000 + " ms");}
}

内存模型与可见性

1. Java内存模型（JMM）

Java内存模型定义了线程如何通过内存进行交互，以及数据在何时对其他线程可见。

public class MemoryModelExample {private int normalVariable = 0;private volatile int volatileVariable = 0;// 线程1public void writer() {normalVariable = 1;    // 可能不立即对其他线程可见volatileVariable = 1;  // 立即对其他线程可见，并建立happens-before关系}// 线程2public void reader() {if (volatileVariable == 1) {  // 如果看到volatile变量的写入// 那么一定能看到之前所有的写入操作assert normalVariable == 1; // 这个断言会成功}}
}

2. 内存屏障（Memory Barriers）

内存屏障防止指令重排序并确保内存操作的可见性：

public class MemoryBarrierExample {private int x = 0;private volatile boolean flag = false;// 写线程public void writer() {x = 42;           // 普通写操作flag = true;      // volatile写操作（包含内存屏障）}// 读线程public void reader() {if (flag) {       // volatile读操作（包含内存屏障）int value = x; // 保证能看到x=42}}
}

3. Happens-Before关系

Happens-Before关系保证了内存操作的可见性和有序性：

public class HappensBeforeExample {private final AtomicReference<String> atomicRef = new AtomicReference<>();public void producer() {String data = prepareData(); // 准备数据atomicRef.set(data);         // 原子操作建立happens-before关系}public void consumer() {String data = atomicRef.get(); // 原子操作if (data != null) {processData(data);         // 保证能看到完整的数据}}private String prepareData() { return "prepared data"; }private void processData(String data) { /* 处理数据 */ }
}

常见问题与解决方案

1. ABA问题

ABA问题是指一个值从A变为B，然后又变回A，CAS操作无法检测到这种变化。

// ABA问题示例
public class ABAExample {private AtomicInteger value = new AtomicInteger(100);// 可能出现ABA问题的操作public boolean problematicOperation() {int current = value.get();  // 读取到100// 在这里，其他线程可能将值改为200，然后又改回100// CAS操作仍然会成功，但实际上值已经被修改过return value.compareAndSet(current, current + 50);}
}// 解决ABA问题：使用版本号
public class ABAFreeSolution {private AtomicStampedReference<Integer> stampedRef = new AtomicStampedReference<>(100, 0);public boolean safeOperation() {int[] stampHolder = new int[1];Integer current = stampedRef.get(stampHolder);int currentStamp = stampHolder[0];// 使用版本号进行CAS操作return stampedRef.compareAndSet(current, current + 50, currentStamp, currentStamp + 1);}
}

2. 内存回收问题

在无锁数据结构中，节点的安全回收是一个挑战：

// 危险的内存回收
public class UnsafeNodeReclamation {private AtomicReference<Node> head = new AtomicReference<>();static class Node {volatile int data;volatile Node next;Node(int data) { this.data = data; }}public boolean removeHead() {Node current = head.get();if (current == null) return false;if (head.compareAndSet(current, current.next)) {// 危险：其他线程可能仍在访问current节点current = null; // 这里可能导致其他线程访问已回收的内存return true;}return false;}
}// 安全的内存回收：使用延迟回收
public class SafeNodeReclamation {private AtomicReference<Node> head = new AtomicReference<>();private final Queue<Node> retiredNodes = new ConcurrentLinkedQueue<>();public boolean removeHead() {Node current = head.get();if (current == null) return false;if (head.compareAndSet(current, current.next)) {retiredNodes.offer(current); // 延迟回收tryReclaim(); // 尝试回收旧节点return true;}return false;}private void tryReclaim() {// 实现安全的内存回收策略// 例如：引用计数、epoch-based回收等}
}

3. 活锁问题

多个线程不断重试但无法进展的情况：

public class LivelockSolution {private AtomicInteger value = new AtomicInteger(0);private static final int MAX_RETRIES = 100;public boolean incrementWithBackoff() {int retries = 0;int current, next;do {current = value.get();next = current + 1;if (++retries > MAX_RETRIES) {// 避免无限重试Thread.yield(); // 让出CPU时间retries = 0;}// 指数退避if (retries > 10) {try {Thread.sleep(1L << Math.min(retries - 10, 10));} catch (InterruptedException e) {Thread.currentThread().interrupt();return false;}}} while (!value.compareAndSet(current, next));return true;}
}

无锁数据结构实现

1. 无锁栈（Lock-Free Stack）

public class LockFreeStack<T> {private final AtomicReference<Node<T>> head = new AtomicReference<>();private static class Node<T> {final T data;volatile Node<T> next;Node(T data) {this.data = data;}}public void push(T item) {Node<T> newNode = new Node<>(item);Node<T> currentHead;do {currentHead = head.get();newNode.next = currentHead;} while (!head.compareAndSet(currentHead, newNode));}public T pop() {Node<T> currentHead;Node<T> newHead;do {currentHead = head.get();if (currentHead == null) {return null; // 栈为空}newHead = currentHead.next;} while (!head.compareAndSet(currentHead, newHead));return currentHead.data;}public boolean isEmpty() {return head.get() == null;}
}

2. 无锁队列（Lock-Free Queue）

public class LockFreeQueue<T> {private final AtomicReference<Node<T>> head;private final AtomicReference<Node<T>> tail;private static class Node<T> {volatile T data;volatile AtomicReference<Node<T>> next;Node(T data) {this.data = data;this.next = new AtomicReference<>();}}public LockFreeQueue() {Node<T> dummy = new Node<>(null);head = new AtomicReference<>(dummy);tail = new AtomicReference<>(dummy);}public void enqueue(T item) {Node<T> newNode = new Node<>(item);while (true) {Node<T> last = tail.get();Node<T> next = last.next.get();if (last == tail.get()) { // 检查tail是否被其他线程修改if (next == null) {// tail指向队列的最后一个节点if (last.next.compareAndSet(null, newNode)) {tail.compareAndSet(last, newNode); // 尝试移动tailbreak;}} else {// tail没有指向最后一个节点，帮助其他线程移动tailtail.compareAndSet(last, next);}}}}public T dequeue() {while (true) {Node<T> first = head.get();Node<T> last = tail.get();Node<T> next = first.next.get();if (first == head.get()) { // 检查head是否被其他线程修改if (first == last) {if (next == null) {return null; // 队列为空}// tail落后，帮助移动tailtail.compareAndSet(last, next);} else {T data = next.data;if (head.compareAndSet(first, next)) {return data;}}}}}
}

3. 无锁哈希表

public class LockFreeHashMap<K, V> {private volatile AtomicReferenceArray<Node<K, V>> table;private final AtomicInteger size = new AtomicInteger(0);private static final int DEFAULT_CAPACITY = 16;private static class Node<K, V> {final K key;volatile V value;volatile Node<K, V> next;Node(K key, V value) {this.key = key;this.value = value;}}public LockFreeHashMap() {table = new AtomicReferenceArray<>(DEFAULT_CAPACITY);}public V put(K key, V value) {int hash = hash(key);AtomicReferenceArray<Node<K, V>> currentTable = table;while (true) {int index = hash & (currentTable.length() - 1);Node<K, V> head = currentTable.get(index);// 查找已存在的键Node<K, V> current = head;while (current != null) {if (current.key.equals(key)) {V oldValue = current.value;current.value = value; // volatile写入return oldValue;}current = current.next;}// 插入新节点Node<K, V> newNode = new Node<>(key, value);newNode.next = head;if (currentTable.compareAndSet(index, head, newNode)) {size.incrementAndGet();return null;}// 重试或处理表扩容}}public V get(K key) {int hash = hash(key);AtomicReferenceArray<Node<K, V>> currentTable = table;int index = hash & (currentTable.length() - 1);Node<K, V> current = currentTable.get(index);while (current != null) {if (current.key.equals(key)) {return current.value;}current = current.next;}return null;}private int hash(K key) {return key == null ? 0 : key.hashCode();}
}

性能分析与对比

1. 吞吐量对比

public class PerformanceComparison {private static final int THREAD_COUNT = Runtime.getRuntime().availableProcessors();private static final int OPERATIONS_PER_THREAD = 1_000_000;// 测试不同实现的性能public void comparePerformance() throws InterruptedException {System.out.println("Performance Comparison Results:");// 1. 无锁实现long lockFreeTime = testLockFreeImplementation();System.out.println("Lock-Free: " + lockFreeTime + " ms");// 2. 锁实现long lockedTime = testLockedImplementation();System.out.println("Locked: " + lockedTime + " ms");// 3. 性能提升比例double improvement = (double) lockedTime / lockFreeTime;System.out.println("Performance improvement: " + String.format("%.2fx", improvement));}private long testLockFreeImplementation() throws InterruptedException {LockFreeStack<Integer> stack = new LockFreeStack<>();return measureTime(() -> {Thread[] threads = new Thread[THREAD_COUNT];for (int i = 0; i < THREAD_COUNT; i++) {threads[i] = new Thread(() -> {for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {stack.push(j);stack.pop();}});threads[i].start();}for (Thread thread : threads) {try {thread.join();} catch (InterruptedException e) {Thread.currentThread().interrupt();}}});}private long testLockedImplementation() throws InterruptedException {Stack<Integer> stack = new Stack<>(); // 同步实现return measureTime(() -> {Thread[] threads = new Thread[THREAD_COUNT];for (int i = 0; i < THREAD_COUNT; i++) {threads[i] = new Thread(() -> {for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {synchronized (stack) {stack.push(j);if (!stack.isEmpty()) {stack.pop();}}}});threads[i].start();}for (Thread thread : threads) {try {thread.join();} catch (InterruptedException e) {Thread.currentThread().interrupt();}}});}private long measureTime(Runnable task) {long startTime = System.currentTimeMillis();task.run();return System.currentTimeMillis() - startTime;}
}

2. 延迟分析

public class LatencyAnalysis {private static final int SAMPLES = 10000;public void analyzeLatency() {long[] lockFreeTimes = new long[SAMPLES];long[] lockedTimes = new long[SAMPLES];LockFreeStack<Integer> lockFreeStack = new LockFreeStack<>();Stack<Integer> lockedStack = new Stack<>();// 测试无锁实现的延迟for (int i = 0; i < SAMPLES; i++) {long start = System.nanoTime();lockFreeStack.push(i);lockFreeStack.pop();lockFreeTimes[i] = System.nanoTime() - start;}// 测试锁实现的延迟for (int i = 0; i < SAMPLES; i++) {long start = System.nanoTime();synchronized (lockedStack) {lockedStack.push(i);if (!lockedStack.isEmpty()) {lockedStack.pop();}}lockedTimes[i] = System.nanoTime() - start;}// 分析结果analyzeResults("Lock-Free", lockFreeTimes);analyzeResults("Locked", lockedTimes);}private void analyzeResults(String type, long[] times) {Arrays.sort(times);long min = times[0];long max = times[times.length - 1];long median = times[times.length / 2];long p95 = times[(int) (times.length * 0.95)];long p99 = times[(int) (times.length * 0.99)];System.out.println(type + " Latency Analysis:");System.out.println("  Min: " + min + " ns");System.out.println("  Median: " + median + " ns");System.out.println("  95th percentile: " + p95 + " ns");System.out.println("  99th percentile: " + p99 + " ns");System.out.println("  Max: " + max + " ns");System.out.println();}
}

3. 可扩展性测试

public class ScalabilityTest {private static final int MAX_THREADS = 32;private static final int OPERATIONS_PER_THREAD = 100_000;public void testScalability() {System.out.println("Scalability Test Results:");System.out.println("Threads\tLock-Free(ops/s)\tLocked(ops/s)\tSpeedup");for (int threadCount = 1; threadCount <= MAX_THREADS; threadCount *= 2) {double lockFreeOps = measureThroughput(threadCount, true);double lockedOps = measureThroughput(threadCount, false);double speedup = lockFreeOps / lockedOps;System.out.printf("%d\t%.0f\t\t%.0f\t\t%.2fx%n", threadCount, lockFreeOps, lockedOps, speedup);}}private double measureThroughput(int threadCount, boolean useLockFree) {long startTime = System.currentTimeMillis();Thread[] threads = new Thread[threadCount];if (useLockFree) {LockFreeStack<Integer> stack = new LockFreeStack<>();for (int i = 0; i < threadCount; i++) {threads[i] = new Thread(() -> {for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {stack.push(j);stack.pop();}});}} else {Stack<Integer> stack = new Stack<>();for (int i = 0; i < threadCount; i++) {threads[i] = new Thread(() -> {for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {synchronized (stack) {stack.push(j);if (!stack.isEmpty()) {stack.pop();}}}});}}for (Thread thread : threads) {thread.start();}for (Thread thread : threads) {try {thread.join();} catch (InterruptedException e) {Thread.currentThread().interrupt();}}long endTime = System.currentTimeMillis();long totalOperations = (long) threadCount * OPERATIONS_PER_THREAD * 2; // push + popdouble duration = (endTime - startTime) / 1000.0;return totalOperations / duration;}
}

实际应用场景

1. 高频交易系统

// 高频交易中的无锁订单簿
public class LockFreeOrderBook {private final LockFreeSkipList<Order> buyOrders = new LockFreeSkipList<>();private final LockFreeSkipList<Order> sellOrders = new LockFreeSkipList<>();static class Order implements Comparable<Order> {final String id;final double price;final int quantity;final long timestamp;Order(String id, double price, int quantity) {this.id = id;this.price = price;this.quantity = quantity;this.timestamp = System.nanoTime();}@Overridepublic int compareTo(Order other) {return Double.compare(this.price, other.price);}}public void addBuyOrder(Order order) {buyOrders.add(order);tryMatch();}public void addSellOrder(Order order) {sellOrders.add(order);tryMatch();}private void tryMatch() {// 高效的无锁订单匹配逻辑Order bestBuy = buyOrders.getMax();Order bestSell = sellOrders.getMin();if (bestBuy != null && bestSell != null && bestBuy.price >= bestSell.price) {// 执行交易executeTrade(bestBuy, bestSell);}}private void executeTrade(Order buy, Order sell) {// 原子性地执行交易// 实现细节...}
}

2. 实时数据处理

// 实时数据流处理中的无锁环形缓冲区
public class LockFreeRingBuffer<T> {private final AtomicReferenceArray<T> buffer;private final int capacity;private final int mask;private final AtomicLong writeSequence = new AtomicLong(0);private final AtomicLong readSequence = new AtomicLong(0);public LockFreeRingBuffer(int capacity) {if ((capacity & (capacity - 1)) != 0) {throw new IllegalArgumentException("Capacity must be power of 2");}this.capacity = capacity;this.mask = capacity - 1;this.buffer = new AtomicReferenceArray<>(capacity);}public boolean offer(T item) {long currentWrite = writeSequence.get();long nextWrite = currentWrite + 1;// 检查是否有空间if (nextWrite - readSequence.get() > capacity) {return false; // 缓冲区满}// 尝试获取写入位置if (writeSequence.compareAndSet(currentWrite, nextWrite)) {int index = (int) (currentWrite & mask);buffer.set(index, item);return true;}return false; // 被其他线程抢占}public T poll() {long currentRead = readSequence.get();// 检查是否有数据if (currentRead >= writeSequence.get()) {return null; // 缓冲区空}// 尝试获取读取位置if (readSequence.compareAndSet(currentRead, currentRead + 1)) {int index = (int) (currentRead & mask);T item = buffer.get(index);buffer.set(index, null); // 清理引用return item;}return null; // 被其他线程抢占}
}

3. 缓存系统

// 高性能缓存中的无锁LRU实现
public class LockFreeLRUCache<K, V> {private final int capacity;private final ConcurrentHashMap<K, Node<K, V>> map;private final AtomicReference<Node<K, V>> head = new AtomicReference<>();private final AtomicReference<Node<K, V>> tail = new AtomicReference<>();private final AtomicInteger size = new AtomicInteger(0);static class Node<K, V> {final K key;volatile V value;volatile AtomicReference<Node<K, V>> prev = new AtomicReference<>();volatile AtomicReference<Node<K, V>> next = new AtomicReference<>();Node(K key, V value) {this.key = key;this.value = value;}}public LockFreeLRUCache(int capacity) {this.capacity = capacity;this.map = new ConcurrentHashMap<>(capacity);// 初始化哨兵节点Node<K, V> headSentinel = new Node<>(null, null);Node<K, V> tailSentinel = new Node<>(null, null);headSentinel.next.set(tailSentinel);tailSentinel.prev.set(headSentinel);head.set(headSentinel);tail.set(tailSentinel);}public V get(K key) {Node<K, V> node = map.get(key);if (node != null) {moveToHead(node); // 更新访问顺序return node.value;}return null;}public void put(K key, V value) {Node<K, V> existing = map.get(key);if (existing != null) {existing.value = value;moveToHead(existing);return;}Node<K, V> newNode = new Node<>(key, value);if (size.get() >= capacity) {evictLRU(); // 驱逐最少使用的节点}map.put(key, newNode);addToHead(newNode);size.incrementAndGet();}private void moveToHead(Node<K, V> node) {removeNode(node);addToHead(node);}private void addToHead(Node<K, V> node) {// 无锁地将节点添加到头部Node<K, V> headSentinel = head.get();Node<K, V> firstNode = headSentinel.next.get();node.prev.set(headSentinel);node.next.set(firstNode);headSentinel.next.set(node);firstNode.prev.set(node);}private void removeNode(Node<K, V> node) {// 无锁地移除节点Node<K, V> prevNode = node.prev.get();Node<K, V> nextNode = node.next.get();prevNode.next.set(nextNode);nextNode.prev.set(prevNode);}private void evictLRU() {Node<K, V> tailSentinel = tail.get();Node<K, V> lastNode = tailSentinel.prev.get();if (lastNode != head.get()) { // 不是头哨兵removeNode(lastNode);map.remove(lastNode.key);size.decrementAndGet();}}
}

最佳实践与注意事项

1. 设计原则

// 1. 最小化共享状态
public class MinimalSharedState {// 好的设计：每个线程有自己的本地状态private static final ThreadLocal<Random> localRandom = ThreadLocal.withInitial(Random::new);private final AtomicLong globalCounter = new AtomicLong(0);public long generateUniqueId() {// 组合线程本地状态和全局状态long threadId = Thread.currentThread().getId();long localValue = localRandom.get().nextInt(1000);long globalValue = globalCounter.incrementAndGet();return (threadId << 32) | (globalValue << 16) | localValue;}
}// 2. 避免复杂的CAS循环
public class SimpleCASOperations {private final AtomicReference<Node> head = new AtomicReference<>();// 好的做法：简单的CAS操作public void addNode(Node newNode) {Node currentHead;do {currentHead = head.get();newNode.next = currentHead;} while (!head.compareAndSet(currentHead, newNode));}// 避免：复杂的多步骤CAS操作// 这样的代码容易出错且难以维护
}

2. 性能优化技巧

public class PerformanceOptimizations {// 1. 使用适当的数据对齐@sun.misc.Contended  // 避免伪共享private volatile long counter1;@sun.misc.Contendedprivate volatile long counter2;// 2. 批量操作减少竞争private final AtomicLong batchCounter = new AtomicLong(0);private static final int BATCH_SIZE = 100;public void batchIncrement() {// 每个线程批量获取计数器值long batchStart = batchCounter.getAndAdd(BATCH_SIZE);// 在本地处理批量操作for (int i = 0; i < BATCH_SIZE; i++) {// 处理 batchStart + i}}// 3. 使用合适的退避策略public boolean incrementWithBackoff(AtomicInteger counter) {int attempts = 0;int current, next;do {current = counter.get();next = current + 1;if (++attempts > 10) {// 指数退避try {Thread.sleep(1L << Math.min(attempts - 10, 6));} catch (InterruptedException e) {Thread.currentThread().interrupt();return false;}}} while (!counter.compareAndSet(current, next));return true;}
}

3. 调试和测试策略

public class LockFreeDebugging {// 1. 添加统计信息private final AtomicLong successfulCAS = new AtomicLong(0);private final AtomicLong failedCAS = new AtomicLong(0);private final AtomicInteger maxRetries = new AtomicInteger(0);public boolean incrementWithStats(AtomicInteger counter) {int retries = 0;int current, next;do {current = counter.get();next = current + 1;retries++;} while (!counter.compareAndSet(current, next));// 更新统计信息successfulCAS.incrementAndGet();int currentMax = maxRetries.get();while (retries > currentMax && !maxRetries.compareAndSet(currentMax, retries)) {currentMax = maxRetries.get();}return true;}// 2. 压力测试public void stressTest() throws InterruptedException {final int THREAD_COUNT = Runtime.getRuntime().availableProcessors() * 2;final int OPERATIONS_PER_THREAD = 1_000_000;AtomicInteger counter = new AtomicInteger(0);Thread[] threads = new Thread[THREAD_COUNT];long startTime = System.currentTimeMillis();for (int i = 0; i < THREAD_COUNT; i++) {threads[i] = new Thread(() -> {for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {incrementWithStats(counter);}});threads[i].start();}for (Thread thread : threads) {thread.join();}long endTime = System.currentTimeMillis();// 验证结果int expectedValue = THREAD_COUNT * OPERATIONS_PER_THREAD;int actualValue = counter.get();System.out.println("Expected: " + expectedValue);System.out.println("Actual: " + actualValue);System.out.println("Correct: " + (expectedValue == actualValue));System.out.println("Time: " + (endTime - startTime) + " ms");System.out.println("Max retries: " + maxRetries.get());}
}

4. 常见陷阱与避免方法

public class CommonPitfalls {// 陷阱1：忘记处理CAS失败public void badIncrement(AtomicInteger counter) {int current = counter.get();int next = current + 1;counter.compareAndSet(current, next); // 错误：没有处理失败情况}public void goodIncrement(AtomicInteger counter) {int current, next;do {current = counter.get();next = current + 1;} while (!counter.compareAndSet(current, next)); // 正确：循环直到成功}// 陷阱2：ABA问题private AtomicReference<Node> head = new AtomicReference<>();public void problematicRemove() {Node current = head.get();if (current != null) {Node next = current.next;// 在这里，其他线程可能修改了链表结构head.compareAndSet(current, next); // 可能出现ABA问题}}// 解决方案：使用版本号private AtomicStampedReference<Node> stampedHead = new AtomicStampedReference<>(null, 0);public void safeRemove() {int[] stampHolder = new int[1];Node current = stampedHead.get(stampHolder);int currentStamp = stampHolder[0];if (current != null) {Node next = current.next;stampedHead.compareAndSet(current, next, currentStamp, currentStamp + 1);}}// 陷阱3：内存可见性问题private int normalField; // 错误：可能不可见private volatile int volatileField; // 正确：保证可见性// 陷阱4：过度使用无锁编程// 不是所有场景都适合无锁编程// 简单的同步操作使用锁可能更合适private final Object simpleLock = new Object();private int simpleCounter = 0;public void simpleIncrement() {synchronized (simpleLock) {simpleCounter++; // 对于简单操作，锁可能更清晰}}
}