Java:实现使用 Treiber 算法的非阻塞堆栈（附带源码）
一、项目背景详细介绍
在高并发、多线程场景下，基于锁的同步结构如 synchronized 或 ReentrantLock 往往成为性能瓶颈，导致线程阻塞和上下文切换开销。无锁（Lock‑Free）数据结构通过原子操作（如 CAS）实现线程间协作，能够显著提升并发吞吐量并降低延迟。Treiber 算法是一种经典的无锁栈（non‑blocking stack）实现，它利用原子比较并交换（CAS）操作在链表头上实现入栈（push）和出栈（pop），在 Java 中可以通过 java.util.concurrent.atomic.AtomicReference 轻松实现。
二、项目需求详细介绍
-
基本栈接口
-
定义泛型接口
NonBlockingStack<E>，提供void push(E item)、E pop()、boolean isEmpty()方法；
-
-
Treiber 算法实现
-
提供类
TreiberStack<E>实现上述接口，内部使用AtomicReference<Node<E>> head； -
push：构造新节点，将其next指向当前head，通过 CAS 更新head； -
pop：读取当前head，若非空，则将headCAS 更新为head.next，返回弹出值；
-
-
高并发安全性
-
所有操作必须无锁且线程安全，利用循环重试保证在 CAS 失败时重试；
-
-
内存管理与 ABA 问题
-
简单实现可忽略 ABA，但需在文档中说明；
-
可选扩展：使用带版本号的
AtomicStampedReference或AtomicMarkableReference解决 ABA；
-
-
性能测试
-
提供基准测试
TreiberStackBenchmark，对比不同并发线程下的吞吐量；
-
-
命令行演示
-
提供
Main类，接受参数：-
--threads t：并发线程数； -
--ops n：每线程操作次数； -
--mode push|pop：测试入栈或出栈吞吐；
-
-
-
单元测试覆盖
-
使用 JUnit 5 对单线程和多线程场景测试：正确性、空栈
pop返回null或抛异常；
-
-
可扩展性
-
可替换节点引用类型支持 ABA 防护；
-
可改造为无锁队列（Michael‑Scott 队列）等其他数据结构。
-
三、相关技术详细介绍
-
原子引用
-
java.util.concurrent.atomic.AtomicReference<T>：支持 CAS 操作； -
AtomicStampedReference<T>：带版本号避免 ABA；
-
-
无锁算法思想
-
CAS 循环重试（spin）机制；
-
ABA 问题及其基本解决方案；
-
-
并发性能基准
-
使用 JMH（Java Microbenchmark Harness）测量吞吐和延迟；
-
比较不同线程数、不同操作比例的表现；
-
-
单元测试与多线程测试
-
JUnit 5 结合
ExecutorService模拟高并发环境； -
使用
CountDownLatch或CyclicBarrier统一启动。
-
-
命令行解析
-
Apache Commons CLI：解析并发测试参数并打印结果。
-
四、实现思路详细介绍
-
节点定义
static class Node<E> {
final E item;
final Node<E> next;
Node(E item, Node<E> next) { this.item = item; this.next = next; }
}
-
Push 操作
-
循环读取
head； -
构造新
Node指向旧head； -
if (head.compareAndSet(old, newNode)) return;否则重试。
-
-
Pop 操作
-
循环读取
head； -
若
old == null返回null； -
if (head.compareAndSet(old, old.next)) return old.item;否则重试。
-
-
Empty 检查
-
直接
return head.get() == null;。
-
-
ABA 防护（可选）
-
将
AtomicReference<Node<E>> head换成AtomicStampedReference<Node<E>>，在 CAS 时携带和比较版本戳；
-
-
性能基准与命令行演示
-
使用 JMH 编写
TreiberStackBenchmark，并在Main中通过 CLI 参数启动特定并发测试。
-
五、完整实现代码
// 文件：src/main/java/com/example/stack/NonBlockingStack.java
package com.example.stack;
/**
* 非阻塞栈接口，提供基本操作
*/
public interface NonBlockingStack<E> {
void push(E item);
E pop();
boolean isEmpty();
}
// 文件：src/main/java/com/example/stack/TreiberStack.java
package com.example.stack;
import java.util.concurrent.atomic.AtomicReference;
/**
* Treiber 无锁栈实现，基于 CAS 操作
*/
public class TreiberStack<E> implements NonBlockingStack<E> {
private static class Node<E> {
final E item;
final Node<E> next;
Node(E item, Node<E> next) {
this.item = item;
this.next = next;
}
}
private final AtomicReference<Node<E>> head = new AtomicReference<>();
@Override
public void push(E item) {
Node<E> newNode;
Node<E> oldHead;
do {
oldHead = head.get();
newNode = new Node<>(item, oldHead);
} while (!head.compareAndSet(oldHead, newNode));
}
@Override
public E pop() {
Node<E> oldHead;
Node<E> next;
do {
oldHead = head.get();
if (oldHead == null) {
return null;
}
next = oldHead.next;
} while (!head.compareAndSet(oldHead, next));
return oldHead.item;
}
@Override
public boolean isEmpty() {
return head.get() == null;
}
}
// 文件：src/main/java/com/example/benchmark/TreiberStackBenchmark.java
package com.example.benchmark;
import com.example.stack.TreiberStack;
import com.example.stack.NonBlockingStack;
import org.openjdk.jmh.annotations.*;
import java.util.concurrent.TimeUnit;
/**
* JMH 基准测试：TreiberStack 在不同并发下的吞吐
*/
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@State(Scope.Benchmark)
public class TreiberStackBenchmark {
@Param({"1","2","4","8","16"})
public int threads;
@Param({"10000","100000"})
public int ops;
private NonBlockingStack<Integer> stack;
@Setup(Level.Iteration)
public void setup() {
stack = new TreiberStack<>();
}
@Benchmark
@Threads(value = 0) // 使用 threads 参数
public void benchPush() {
for (int i = 0; i < ops; i++) {
stack.push(i);
}
}
@Benchmark
@Threads(value = 0)
public void benchPop() {
for (int i = 0; i < ops; i++) {
stack.pop();
}
}
}
// 文件：src/main/java/com/example/cli/Main.java
package com.example.cli;
import com.example.stack.TreiberStack;
import com.example.stack.NonBlockingStack;
import org.apache.commons.cli.*;
import java.util.concurrent.*;
/**
* Main 类通过命令行演示 TreiberStack 并发吞吐测试
*/
public class Main {
public static void main(String[] args) {
Options opts = new Options();
opts.addOption(null, "threads", true, "线程数");
opts.addOption(null, "ops", true, "每线程操作次数");
opts.addOption(null, "mode", true, "测试模式：push|pop");
HelpFormatter hf = new HelpFormatter();
CommandLineParser parser = new DefaultParser();
try {
CommandLine cmd = parser.parse(opts, args);
int threads = Integer.parseInt(cmd.getOptionValue("threads","4"));
int ops = Integer.parseInt(cmd.getOptionValue("ops","100000"));
String mode = cmd.getOptionValue("mode","push");
NonBlockingStack<Integer> stack = new TreiberStack<>();
ExecutorService pool = Executors.newFixedThreadPool(threads);
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(threads);
for (int t = 0; t < threads; t++) {
pool.submit(() -> {
try {
start.await();
if ("pop".equalsIgnoreCase(mode)) {
for (int i = 0; i < ops; i++) stack.pop();
} else {
for (int i = 0; i < ops; i++) stack.push(i);
}
} catch (InterruptedException ignored) {
} finally {
done.countDown();
}
});
}
long begin = System.nanoTime();
start.countDown();
done.await();
long duration = System.nanoTime() - begin;
pool.shutdown();
System.out.printf("模式=%s, 线程=%d, 操作次数/线程=%d, 总耗时=%.3fms%n",
mode, threads, ops, duration/1_000_000.0);
} catch (Exception e) {
System.err.println("参数解析或执行错误: " + e.getMessage());
hf.printHelp("java -jar treiber-stack-demo.jar", opts);
}
}
}
// 文件：src/test/java/com/example/stack/TreiberStackTest.java
package com.example.stack;
import org.junit.jupiter.api.Test;
import java.util.concurrent.*;
import java.util.Set;
import java.util.HashSet;
import static org.junit.jupiter.api.Assertions.*;
/**
* TreiberStack 单元测试
*/
public class TreiberStackTest {
@Test
void testSingleThreadPushPop() {
NonBlockingStack<Integer> stack = new TreiberStack<>();
assertTrue(stack.isEmpty());
stack.push(1);
stack.push(2);
assertFalse(stack.isEmpty());
assertEquals(2, stack.pop());
assertEquals(1, stack.pop());
assertNull(stack.pop());
assertTrue(stack.isEmpty());
}
@Test
void testMultiThreadPush() throws InterruptedException {
NonBlockingStack<Integer> stack = new TreiberStack<>();
int threads = 4, ops = 10000;
ExecutorService pool = Executors.newFixedThreadPool(threads);
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(threads);
for (int t = 0; t < threads; t++) {
int base = t * ops;
pool.submit(() -> {
try {
start.await();
for (int i = 0; i < ops; i++) stack.push(base + i);
} catch (InterruptedException ignored) {
} finally {
done.countDown();
}
});
}
start.countDown();
done.await();
pool.shutdown();
Set<Integer> seen = new HashSet<>();
Integer v;
while ((v = stack.pop()) != null) {
seen.add(v);
}
assertEquals(threads * ops, seen.size());
}
}
六、代码详细解读
-
接口 NonBlockingStack<E>
定义无锁栈的基本操作：push、pop、isEmpty，解耦实现。 -
TreiberStack<E>
-
使用
AtomicReference<Node<E>> head保存栈顶； -
push：循环读取当前
head，构造新节点指向旧head，利用compareAndSet更新； -
pop：循环读取
head，若非空则尝试 CAS 将head指向下一个节点，成功后返回弹出元素； -
CAS 失败时自动重试，无锁且线程安全。
-
-
JMH 基准 TreiberStackBenchmark
-
参数化线程数和操作次数；
-
分别测试
push与pop操作的吞吐； -
@Threads(value=0)结合@Param自适应并发度。
-
-
命令行 Main
-
使用
CountDownLatch同步所有线程起跑和完成； -
记录纳秒级起止时间，统计并发模式下的总耗时；
-
支持
--mode push|pop，展示不同操作性能。
-
-
单元测试 TreiberStackTest
-
单线程：测试基本功能和空栈
pop返回null； -
多线程：4 线程并发 push，各线程推送唯一区间数值，最后 pop 集合，验证无丢失或重复。
-
七、项目详细总结
本项目基于 Treiber 算法实现了无锁栈 TreiberStack<E>，利用 Java 原子引用（CAS）完成线程安全的并发 push 和 pop 操作。
-
无锁设计：避免线程阻塞和上下文切换，提高高并发场景下的吞吐和可伸缩性。
-
可测试性：单元测试覆盖单线程与多线程正确性，JMH 基准测试量化性能。
-
可演示性：命令行工具直观展示不同线程数和操作类型下的耗时。
八、项目常见问题及解答
Q1：为何不需锁就能保证线程安全？
CAS 原子操作确保只有一个线程能成功更新 head，失败者自动重试，无需显式锁。
Q2：出现 ABA 问题会如何？
若节点 A 出栈后再入栈，head 从 A 到 B 再回到 A，CAS 可能误判成功。此实现未防 ABA，文档中需说明可用 AtomicStampedReference 版本解决。
Q3：pop 返回 null 合适吗？
本实现选择空栈时返回 null；可根据需求改为抛出异常。
Q4：如何避免无限自旋？
CAS 重试在高竞争时可能长时间循环，可在重试中插入 Thread.yield() 或退避策略。
Q5：此栈能保证 FIFO 吗？
栈为 LIFO 结构，若需要 FIFO，请实现无锁队列（如 Michael‑Scott 算法）。
九、扩展方向与性能优化
-
ABA 防护
使用AtomicStampedReference<Node<E>> head，在 CAS 时携带版本戳，避免 ABA。 -
迭代版优化
将递归或深度自旋替换为限次重试策略或带退避的自旋，提升性能稳定性。 -
无锁队列
基于 Michael‑Scott 算法实现无锁队列（NonBlockingQueue），用于 FIFO 场景。 -
内存屏障与缓存
深入研究 Java 内存模型中 CAS 的性能，并结合缓存行对齐减少伪共享。 -
基准对比
使用 JMH 对比TreiberStack与传统synchronized、ReentrantLock的性能差异，指导生产环境选型。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
14
0- 0
已为社区贡献484条内容
所有评论(0)