Java:实现CAS 的非阻塞算法（附带源码）
一、项目背景详细介绍
在高并发系统中，传统的锁（如 synchronized、ReentrantLock）可能导致线程阻塞、上下文切换、死锁等问题，成为性能瓶颈。无锁（Lock‑Free）算法利用硬件原子指令（如 CAS，Compare-And-Set）实现并发数据结构与算法，能显著提升吞吐量并降低延迟。CAS 是最基础的原子操作，通过比较内存中的当前值与期望值是否相同来决定是否更新，常见于 java.util.concurrent.atomic 包中的原子引用与数值类型。掌握 CAS 的应用是实现更高层无锁算法（如无锁栈、无锁队列、无锁计数器）的基础。
二、项目需求详细介绍
本项目要求在 Java 中演示 CAS 的典型无锁算法实现，具体包括：
-
无锁计数器（Atomic Counter）
-
提供类
CasCounter，内部用原子字段实现递增和读取：-
void increment() -
long get()
-
-
-
自旋锁（Spin Lock）
-
使用
AtomicBoolean或AtomicReference<Thread>实现自旋锁，提供：-
void lock()自旋获取锁 -
void unlock()释放锁
-
-
-
无锁队列（Michael‑Scott 队列）
-
实现
NonBlockingQueue<E>接口，提供enqueue(E)、E dequeue()、boolean isEmpty()方法； -
基于 CAS 操作在链表尾部/头部进行原子更新，保证线程安全。
-
-
性能对比
-
与基于
ReentrantLock的相同功能实现进行吞吐和延迟对比测试；
-
-
单元测试覆盖
-
使用 JUnit 5 对以上三种算法在单线程和多线程场景下测试正确性与并发安全；
-
-
命令行演示
-
提供
Main类，通过 CLI 选择算法类型（counter、spinlock、queue）与线程数及操作次数，统计运行耗时与吞吐。
-
三、相关技术详细介绍
-
CAS 原子操作
-
AtomicLong.compareAndSet(expected, update) -
AtomicReference<V>.compareAndSet(expected, update)
-
-
自旋机制
-
在失败时不断重试，不阻塞线程；
-
可结合
LockSupport.parkNanos做轻微退避。
-
-
Michael‑Scott 无锁队列
-
经典无锁队列，使用双原子引用
head和tail； -
入队：CAS 更新
tail.next并推进tail； -
出队：CAS 读取
head.next并推进head；
-
-
线程与同步
-
ExecutorService模拟并发环境； -
CountDownLatch协调启动与结束；
-
-
性能基准
-
可选 JMH 进行微基准，亦可在
Main中简单统计系统时间。
-
四、实现思路详细介绍
-
CasCounter
-
内部持有
AtomicLong value = new AtomicLong(0)； -
increment()调用value.incrementAndGet()或手写 CAS 循环：
-
long prev, next;
do {
prev = value.get();
next = prev + 1;
} while (!value.compareAndSet(prev, next));
SpinLock
-
使用
AtomicBoolean locked = new AtomicBoolean(false)； -
lock()：
while (!locked.compareAndSet(false, true)) {
// 自旋：可选择 Thread.yield() 或 parkNanos
}
-
-
unlock()：locked.set(false);
-
-
NonBlockingQueue<E>
-
定义静态内部类
Node<E>，包含item和AtomicReference<Node<E>> next； -
head和tail都为AtomicReference<Node<E>>，初始化时指向哨兵节点； -
enqueue(E)：创建新节点，循环读取tail，尝试tail.get().next.compareAndSet(null, newNode)，成功后再tail.compareAndSet(oldTail, newNode)； -
dequeue()：循环读取head，检查head.get().next是否为空，若非空则尝试head.compareAndSet(oldHead, nextNode)并返回nextNode.item。
-
-
命令行 Main
-
解析
--algo counter|spinlock|queue、--threads、--ops； -
根据算法类型构造对应实例并在多线程中交替调用操作；
-
记录开始和结束时间，输出总耗时及平均每操作耗时。
-
五、完整实现代码
// 文件：src/main/java/com/example/cas/CasCounter.java
package com.example.cas;
import java.util.concurrent.atomic.AtomicLong;
/**
* CasCounter：基于 CAS 的无锁计数器
*/
public class CasCounter {
private final AtomicLong value = new AtomicLong(0);
/**
* 原子递增
*/
public void increment() {
long prev, next;
do {
prev = value.get();
next = prev + 1;
} while (!value.compareAndSet(prev, next));
}
/**
* 获取当前值
*/
public long get() {
return value.get();
}
}
// 文件：src/main/java/com/example/lock/SpinLock.java
package com.example.lock;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* SpinLock：简单自旋锁实现
*/
public class SpinLock {
private final AtomicBoolean locked = new AtomicBoolean(false);
/**
* 获取锁：自旋直到成功
*/
public void lock() {
while (!locked.compareAndSet(false, true)) {
// 可以选择 Thread.yield() 或短暂休眠以降低 CPU 占用
}
}
/**
* 释放锁
*/
public void unlock() {
locked.set(false);
}
}
// 文件：src/main/java/com/example/queue/NonBlockingQueue.java
package com.example.queue;
/**
* 无锁队列接口
*/
public interface NonBlockingQueue<E> {
void enqueue(E item);
E dequeue();
boolean isEmpty();
}
// 文件：src/main/java/com/example/queue/MichaelScottQueue.java
package com.example.queue;
import java.util.concurrent.atomic.AtomicReference;
/**
* Michael‑Scott 无锁队列实现
*/
public class MichaelScottQueue<E> implements NonBlockingQueue<E> {
private static class Node<E> {
final E item;
final AtomicReference<Node<E>> next;
Node(E item) {
this.item = item;
this.next = new AtomicReference<>(null);
}
}
private final AtomicReference<Node<E>> head;
private final AtomicReference<Node<E>> tail;
public MichaelScottQueue() {
Node<E> dummy = new Node<>(null);
head = new AtomicReference<>(dummy);
tail = new AtomicReference<>(dummy);
}
@Override
public void enqueue(E item) {
Node<E> node = new Node<>(item);
while (true) {
Node<E> t = tail.get();
Node<E> next = t.next.get();
if (t == tail.get()) {
if (next == null) {
if (t.next.compareAndSet(null, node)) {
tail.compareAndSet(t, node);
return;
}
} else {
tail.compareAndSet(t, next);
}
}
}
}
@Override
public E dequeue() {
while (true) {
Node<E> h = head.get();
Node<E> t = tail.get();
Node<E> next = h.next.get();
if (h == head.get()) {
if (h == t) {
if (next == null) {
return null;
}
tail.compareAndSet(t, next);
} else {
E value = next.item;
if (head.compareAndSet(h, next)) {
return value;
}
}
}
}
}
@Override
public boolean isEmpty() {
return head.get().next.get() == null;
}
}
// 文件：src/main/java/com/example/cli/Main.java
package com.example.cli;
import com.example.cas.CasCounter;
import com.example.lock.SpinLock;
import com.example.queue.MichaelScottQueue;
import com.example.queue.NonBlockingQueue;
import org.apache.commons.cli.*;
import java.util.concurrent.*;
/**
* Main：命令行演示 CAS 算法、SpinLock、无锁队列性能
*/
public class Main {
public static void main(String[] args) {
Options opts = new Options();
opts.addOption(null, "algo", true, "算法：counter|spinlock|queue");
opts.addOption(null, "threads",true, "线程数，默认4");
opts.addOption(null, "ops", true, "每线程操作次数，默认100000");
HelpFormatter hf = new HelpFormatter();
CommandLineParser parser = new DefaultParser();
try {
CommandLine cmd = parser.parse(opts, args);
String algo = cmd.getOptionValue("algo", "counter");
int threads = Integer.parseInt(cmd.getOptionValue("threads", "4"));
int ops = Integer.parseInt(cmd.getOptionValue("ops", "100000"));
ExecutorService pool = Executors.newFixedThreadPool(threads);
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(threads);
long startNanos, duration;
switch (algo) {
case "spinlock":
SpinLock lock = new SpinLock();
startNanos = System.nanoTime();
for (int t = 0; t < threads; t++) {
pool.submit(() -> {
try {
start.await();
for (int i = 0; i < ops; i++) {
lock.lock();
lock.unlock();
}
} catch (InterruptedException ignored) {}
finally { done.countDown(); }
});
}
break;
case "queue":
NonBlockingQueue<Integer> queue = new MichaelScottQueue<>();
startNanos = System.nanoTime();
for (int t = 0; t < threads; t++) {
pool.submit(() -> {
try {
start.await();
for (int i = 0; i < ops; i++) {
queue.enqueue(i);
queue.dequeue();
}
} catch (InterruptedException ignored) {}
finally { done.countDown(); }
});
}
break;
default: // counter
CasCounter counter = new CasCounter();
startNanos = System.nanoTime();
for (int t = 0; t < threads; t++) {
pool.submit(() -> {
try {
start.await();
for (int i = 0; i < ops; i++) {
counter.increment();
}
} catch (InterruptedException ignored) {}
finally { done.countDown(); }
});
}
}
start.countDown();
done.await();
duration = System.nanoTime() - startNanos;
pool.shutdown();
System.out.printf("算法=%s, 线程=%d, 每线程操作=%d, 总耗时=%.3fms%n",
algo, threads, ops, duration/1_000_000.0);
} catch (Exception e) {
System.err.println("错误: " + e.getMessage());
hf.printHelp("java -jar cas-demo.jar", opts);
}
}
}
// 文件：src/test/java/com/example/cas/TreiberCasTest.java
package com.example.cas;
import com.example.cas.CasCounter;
import com.example.lock.SpinLock;
import com.example.queue.MichaelScottQueue;
import com.example.queue.NonBlockingQueue;
import org.junit.jupiter.api.Test;
import java.util.concurrent.*;
import java.util.HashSet;
import java.util.Set;
import static org.junit.jupiter.api.Assertions.*;
/**
* 单元测试：CasCounter、SpinLock、MichaelScottQueue
*/
public class TreiberCasTest {
@Test
void testCasCounterSingle() {
CasCounter c = new CasCounter();
assertEquals(0, c.get());
c.increment();
assertEquals(1, c.get());
}
@Test
void testCasCounterMulti() throws InterruptedException {
CasCounter c = new CasCounter();
int threads = 4, ops = 10000;
ExecutorService pool = Executors.newFixedThreadPool(threads);
CountDownLatch start=new CountDownLatch(1), done=new CountDownLatch(threads);
for (int i=0;i<threads;i++){
pool.submit(() -> {
try {
start.await();
for(int j=0;j<ops;j++) c.increment();
} catch (InterruptedException ignored) {}
finally { done.countDown(); }
});
}
start.countDown(); done.await(); pool.shutdown();
assertEquals(threads*ops, c.get());
}
@Test
void testSpinLock() throws InterruptedException {
SpinLock lock = new SpinLock();
assertDoesNotThrow(() -> {
lock.lock();
lock.unlock();
});
}
@Test
void testQueueMulti() throws InterruptedException {
NonBlockingQueue<Integer> q = new MichaelScottQueue<>();
int threads=4, ops=10000;
ExecutorService pool=Executors.newFixedThreadPool(threads);
CountDownLatch start=new CountDownLatch(1), done=new CountDownLatch(threads);
for(int t=0;t<threads;t++){
pool.submit(() -> {
try {
start.await();
for(int i=0;i<ops;i++){
q.enqueue(i);
q.dequeue();
}
} catch (InterruptedException ignored) {}
finally { done.countDown(); }
});
}
start.countDown(); done.await(); pool.shutdown();
assertTrue(q.isEmpty());
}
}
六、代码详细解读
-
CasCounter
使用AtomicLong的 CAS 循环自增，保证increment()无锁且线程安全。 -
SpinLock
基于AtomicBoolean，lock()不断compareAndSet(false, true)自旋获取锁，unlock()简单set(false)。 -
MichaelScottQueue
双原子引用head和tail，使用经典 Michael‑Scott 算法：-
enqueue：CAS 更新
tail.next，再推进tail； -
dequeue：CAS 更新
head到下一个节点并返回。
-
-
Main
通过 CLI 选定算法，同时启动 N 线程，每线程 O 次操作，并使用CountDownLatch同步启动与结束，记录并行总耗时。 -
单元测试
-
CasCounter：单线程和多线程累加测试；
-
SpinLock：简单 lock/unlock 测试无异常；
-
MichaelScottQueue：并发 enqueue/dequeue 后队列应为空。
-
七、项目详细总结
本项目通过 Java 原子类和 CAS 操作演示了三种无锁并发算法：
-
无锁计数器：高吞吐的线程安全自增。
-
自旋锁：轻量级锁，用于短临界区场景。
-
无锁队列：经典 Michael‑Scott 算法实现的并发队列，可用于消息传递与任务调度。
结合 JUnit 测试和命令行性能演示，可直观了解无锁算法在多线程环境下的正确性与性能优势。
八、项目常见问题及解答
Q1：CAS 自旋会导致 CPU 飙高？
是的，高争用时会不断自旋，建议在自旋中加入退避（Thread.yield() 或 parkNanos）以降低 CPU 占用。
Q2：ABA 问题如何解决？
可使用 AtomicStampedReference 或 AtomicMarkableReference 在 CAS 中携带版本号，避免节点被快速复用误判。
Q3：SpinLock 与 ReentrantLock 性能对比？
SpinLock 无线程挂起开销，但高争用时会浪费 CPU；ReentrantLock 会挂起线程，适合长临界区。
Q4：Michael‑Scott 队列能保证 FIFO 吗？
能，算法设计保证了入队顺序与出队顺序一致。
Q5：何时使用无锁结构？
在高并发、低延迟且操作粒度小的场景，无锁结构能显著提升性能，避免锁竞争瓶颈。
九、扩展方向与性能优化
-
退避策略
在自旋循环中加入指数退避，降低高争用时的自旋成本。 -
ABA 防护
将AtomicReference改为AtomicStampedReference版本，携带版本号解决 ABA。 -
无锁栈
基于 Treiber 算法实现无锁栈（前文已有实现），与无锁队列互补。 -
无锁集合
实现无锁链表、无锁哈希表等更复杂数据结构。 -
JMH 基准
使用 JMH 对比无锁结构与加锁结构在不同线程数、不同负载下的性能差异，指导选型。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
18
0- 0
已为社区贡献484条内容
所有评论(0)