Skip to content

Commit 4e0b6e1

Browse files
Java-EdgeJava-Edge
committed
更新CAS
1 parent 543d060 commit 4e0b6e1

File tree

10 files changed

+406
-0
lines changed

10 files changed

+406
-0
lines changed

‎concurrency/src/main/java/com/javaedge/concurrency/cas/demo/ConcurrentStack.java‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
package com.study.lock.cas.demo;
2+
3+
import java.util.concurrent.TimeUnit;
4+
import java.util.concurrent.atomic.AtomicStampedReference;
5+
6+
public class ConcurrentStack {
7+
AtomicStampedReference<Node> top = new AtomicStampedReference<Node>(null,0);
8+
public void push(Node node){
9+
Node oldTop;
10+
int v;
11+
do{
12+
v=top.getStamp();
13+
oldTop = top.getReference();
14+
node.next = oldTop;
15+
}
16+
while(!top.compareAndSet(oldTop, node,v,v+1));
17+
// }while(!top.compareAndSet(oldTop, node,top.getStamp(),top.getStamp()+1));
18+
}
19+
public Node pop(int time){
20+
Node newTop;
21+
Node oldTop;
22+
int v;
23+
do{
24+
v=top.getStamp();
25+
oldTop = top.getReference();
26+
if(oldTop == null){
27+
return null;
28+
}
29+
newTop = oldTop.next;
30+
try {
31+
if (time != 0) {
32+
System.out.println(Thread.currentThread() + " 睡一下，预期拿到的数据" + oldTop.item);
33+
TimeUnit.SECONDS.sleep(time); // 休眠指定的时间
34+
}
35+
}
36+
catch (InterruptedException e) {
37+
e.printStackTrace();
38+
}
39+
}
40+
while(!top.compareAndSet(oldTop, newTop,v,v+1));
41+
// }while(!top.compareAndSet(oldTop, newTop,top.getStamp(),top.getStamp()));
42+
return oldTop;
43+
}
44+
public void get(){
45+
Node node = top.getReference();
46+
while(node!=null){
47+
System.out.println(node.item);
48+
node = node.next;
49+
}
50+
}
51+
}

‎concurrency/src/main/java/com/javaedge/concurrency/cas/demo/Node.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
package com.javaedge.concurrency.cas.demo;
2+
3+
/**
4+
* 存储在栈里面元素 -- 对象
5+
* @author JavaEdge
6+
* @date 2019/10/18
7+
*/
8+
public class Node {
9+
public final String item;
10+
public Node next;
11+
12+
public Node(String item) {
13+
this.item = item;
14+
}
15+
16+
@Override
17+
public String toString() {
18+
return "item内容:" + this.item;
19+
}
20+
}

‎concurrency/src/main/java/com/javaedge/concurrency/cas/demo/Stack.java‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
package com.javaedge.concurrency.cas.demo;
2+
3+
4+
import java.util.concurrent.TimeUnit;
5+
import java.util.concurrent.atomic.AtomicReference;
6+
7+
/**
8+
* 实现一个 栈（后进先出）
9+
*
10+
* @author JavaEdge
11+
* @date 2019/10/20
12+
*/
13+
public class Stack {
14+
15+
/**
16+
* top cas无锁修改
17+
*/
18+
AtomicReference<Node> top = new AtomicReference<>();
19+
20+
public void push(Node node) { // 入栈
21+
Node oldTop;
22+
do {
23+
oldTop = top.get();
24+
node.next = oldTop;
25+
}
26+
// CAS 替换栈顶
27+
while (!top.compareAndSet(oldTop, node));
28+
}
29+
30+
/**
31+
* 为了演示ABA效果， 增加一个CAS操作的延时
32+
*
33+
* @param time
34+
* @return
35+
* @throws InterruptedException
36+
*/
37+
public Node pop(int time) throws InterruptedException { // 出栈 -- 取出栈顶
38+
39+
Node newTop;
40+
Node oldTop;
41+
do {
42+
oldTop = top.get();
43+
if (oldTop == null) {
44+
return null;
45+
}
46+
newTop = oldTop.next;
47+
if (time != 0) {
48+
System.out.println(Thread.currentThread() + " 睡一下，预期拿到的数据" + oldTop.item);
49+
// 休眠指定的时间
50+
TimeUnit.SECONDS.sleep(time);
51+
}
52+
}
53+
while (!top.compareAndSet(oldTop, newTop));
54+
return oldTop;
55+
}
56+
}

‎concurrency/src/main/java/com/javaedge/concurrency/cas/demo/Test.java‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
package com.study.lock.cas.demo;
2+
3+
public class Test {
4+
public static void main(String[] args) throws InterruptedException {
5+
Stack stack = new Stack();
6+
stack.push(new Node("B"));
7+
stack.push(new Node("A"));
8+
9+
Thread thread1 = new Thread(() -> {
10+
try {
11+
System.out.println(Thread.currentThread() + " 拿到数据：" + stack.pop(3));
12+
// #再继续拿，就会有问题了，理想情况stack出数据应该是 A->C->D->B，实际上ABA问题导致A-B->null
13+
System.out.println(Thread.currentThread() + " 拿到数据：" + stack.pop(0));
14+
System.out.println(Thread.currentThread() + " 拿到数据：" + stack.pop(0));
15+
System.out.println(Thread.currentThread() + " 拿到数据：" + stack.pop(0));
16+
} catch (Exception e) {
17+
e.printStackTrace();
18+
}
19+
});
20+
thread1.start();
21+
22+
Thread.sleep(300); // 让线程1先启动
23+
24+
Thread thread2 = new Thread(() -> {
25+
Node A = null;
26+
try {
27+
A = stack.pop(0);
28+
System.out.println(Thread.currentThread() + " 拿到数据：" + A);
29+
stack.push(new Node("D"));
30+
stack.push(new Node("C"));
31+
stack.push(A);
32+
} catch (Exception e) {
33+
e.printStackTrace();
34+
}
35+
});
36+
thread2.start();
37+
}
38+
}

‎concurrency/src/main/java/com/javaedge/concurrency/example/atomic/AtomicTest.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package com.javaedge.concurrency.example.atomic;
2+
3+
import java.util.concurrent.atomic.AtomicInteger;
4+
5+
/**
6+
* atomic 相关测试代码
7+
*
8+
* @author JavaEdge
9+
* @date 2019/10/20
10+
*/
11+
public class AtomicTest {
12+
13+
public static void main(String[] args) throws InterruptedException {
14+
// 自增
15+
AtomicInteger atomicInteger = new AtomicInteger(0);
16+
for (int i = 0; i < 2; i++) {
17+
new Thread(() -> {
18+
for (int j = 0; j < 10000; j++) {
19+
atomicInteger.incrementAndGet();
20+
}
21+
}).start();
22+
}
23+
Thread.sleep(2000L);
24+
System.out.println(atomicInteger.get());
25+
}
26+
}

‎concurrency/src/main/java/com/javaedge/concurrency/example/atomic/LockCASDemo.java‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
package com.javaedge.concurrency.example.atomic;
2+
3+
import sun.misc.Unsafe;
4+
5+
import java.io.IOException;
6+
import java.lang.reflect.Field;
7+
8+
/**
9+
* 两个线程，对 i 变量进行递增操作
10+
*
11+
* @author JavaEdge
12+
* @date 2019/10/18
13+
*/
14+
public class LockCASDemo {
15+
16+
volatile int i = 0;
17+
18+
private static Unsafe unsafe;
19+
20+
/**
21+
*
22+
* 属性偏移量，用于JVM去定位属性在内存中的地址
23+
*/
24+
static long valueOffset;
25+
26+
static {
27+
try {
28+
Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
29+
theUnsafe.setAccessible(true);
30+
unsafe = (Unsafe) theUnsafe.get(null);
31+
32+
// CAS 硬件原语 ---java语言 无法直接改内存。 曲线通过对象及属性的定位方式
33+
valueOffset = unsafe.objectFieldOffset(LockCASDemo.class.getDeclaredField("i"));
34+
} catch (Exception e) {
35+
e.printStackTrace();
36+
}
37+
// unsafe =
38+
}
39+
40+
public void add() { // 方法栈帧~ 局部变量
41+
// 无锁编程 -- cas + 自旋锁
42+
// TODO xx00
43+
// i++; // 三次操作
44+
int current;
45+
int value;
46+
do {
47+
current = unsafe.getIntVolatile(this, valueOffset); // 读取当前值
48+
value = current + 1; // 计算
49+
} while (!unsafe.compareAndSwapInt(this, valueOffset, current, value));// CAS 底层API
50+
51+
// if(current == i) {
52+
// i = value; // 赋值
53+
// } else {
54+
// // 值发生变化，修改失败
55+
// }
56+
}
57+
58+
public static void main(String[] args) throws InterruptedException, IOException {
59+
LockCASDemo ld = new LockCASDemo();
60+
61+
for (int i = 0; i < 2; i++) { // 2w相加，20000
62+
new Thread(() -> {
63+
for (int j = 0; j < 10000; j++) {
64+
ld.add();
65+
}
66+
}).start();
67+
}
68+
System.in.read(); // 输入任意键退出
69+
System.out.println(ld.i);
70+
}
71+
}
72+

‎concurrency/src/main/java/com/javaedge/concurrency/example/atomic/LongAccumulatorDemo.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
package com.javaedge.concurrency.example.atomic;
2+
3+
import java.util.concurrent.atomic.LongAccumulator;
4+
import java.util.function.LongBinaryOperator;
5+
6+
/**
7+
* LongAdder增强版，处理累加之外，可以自行定义其他计算
8+
*
9+
* @author JavaEdge
10+
* @date 2019/10/18
11+
*/
12+
public class LongAccumulatorDemo {
13+
public static void main(String[] args) throws InterruptedException {
14+
LongAccumulator accumulator = new LongAccumulator(new LongBinaryOperator() {
15+
@Override
16+
public long applyAsLong(long left, long right) {
17+
// 返回最大值，这就是自定义的计算
18+
return left < right ? left : right;
19+
}
20+
}, 0);
21+
22+
// 1000个线程
23+
for (int i = 0; i < 1000; i++) {
24+
int finalI = i;
25+
new Thread(() -> {
26+
accumulator.accumulate(finalI); // 此处实际就是执行上面定义的操作
27+
}).start();
28+
}
29+
30+
Thread.sleep(2000L);
31+
System.out.println(accumulator.longValue()); // 打印出结果
32+
}
33+
34+
}

‎concurrency/src/main/java/com/javaedge/concurrency/example/atomic/LongAdderDemo.java‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
package com.javaedge.concurrency.example.atomic;
2+
3+
import java.util.concurrent.atomic.AtomicLong;
4+
import java.util.concurrent.atomic.LongAdder;
5+
6+
/**
7+
* 测试用例： 同时运行2秒，检查谁的次数最多
8+
*
9+
* @author JavaEdge
10+
* @date 2019/10/18
11+
*/
12+
public class LongAdderDemo {
13+
private long count = 0;
14+
15+
/**
16+
* 同步代码块的方式
17+
* @throws InterruptedException
18+
*/
19+
public void testSync() throws InterruptedException {
20+
for (int i = 0; i < 3; i++) {
21+
new Thread(() -> {
22+
long starttime = System.currentTimeMillis();
23+
while (System.currentTimeMillis() - starttime < 2000) { // 运行两秒
24+
synchronized (this) {
25+
++count;
26+
}
27+
}
28+
long endtime = System.currentTimeMillis();
29+
System.out.println("SyncThread spend:" + (endtime - starttime) + "ms" + " v" + count);
30+
}).start();
31+
}
32+
}
33+
34+
/**
35+
* Atomic方式
36+
*/
37+
private AtomicLong acount = new AtomicLong(0L);
38+
39+
public void testAtomic() throws InterruptedException {
40+
for (int i = 0; i < 3; i++) {
41+
new Thread(() -> {
42+
long starttime = System.currentTimeMillis();
43+
while (System.currentTimeMillis() - starttime < 2000) { // 运行两秒
44+
acount.incrementAndGet(); // acount++;
45+
}
46+
long endtime = System.currentTimeMillis();
47+
System.out.println("AtomicThread spend:" + (endtime - starttime) + "ms" + " v-" + acount.incrementAndGet());
48+
}).start();
49+
}
50+
}
51+
52+
// LongAdder 方式
53+
private LongAdder lacount = new LongAdder();
54+
public void testLongAdder() throws InterruptedException {
55+
for (int i = 0; i < 3; i++) {
56+
new Thread(() -> {
57+
long starttime = System.currentTimeMillis();
58+
while (System.currentTimeMillis() - starttime < 2000) { // 运行两秒
59+
lacount.increment();
60+
}
61+
long endtime = System.currentTimeMillis();
62+
System.out.println("LongAdderThread spend:" + (endtime - starttime) + "ms" + " v-" + lacount.sum());
63+
}).start();
64+
}
65+
}
66+
67+
public static void main(String[] args) throws InterruptedException {
68+
LongAdderDemo demo = new LongAdderDemo();
69+
demo.testSync();
70+
demo.testAtomic();
71+
demo.testLongAdder();
72+
}
73+
}

0 commit comments

Comments
 (0)