剑指JUC原理-9.Java无锁模型

• 👏作者简介：大家好，我是爱吃芝士的土豆倪，24届校招生Java选手，很高兴认识大家
• 📕系列专栏：Spring源码、JUC源码
• 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
• 🍂博主正在努力完成2023计划中：源码溯源，一探究竟
• 📝联系方式：nhs19990716，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀

问题提出

有如下需求，保证 account.withdraw 取款方法的线程安全

import java.util.ArrayList;
import java.util.List;interface Account {// 获取余额Integer getBalance();// 取款void withdraw(Integer amount);/*** 方法内会启动 1000 个线程，每个线程做 -10 元 的操作* 如果初始余额为 10000 那么正确的结果应当是 0*/static void demo(Account account) {List<Thread> ts = new ArrayList<>();long start = System.nanoTime();for (int i = 0; i < 1000; i++) {ts.add(new Thread(() -> {account.withdraw(10);}));}ts.forEach(Thread::start);ts.forEach(t -> {try {t.join();} catch (InterruptedException e) {e.printStackTrace();}});long end = System.nanoTime();System.out.println(account.getBalance()+ " cost: " + (end-start)/1000_000 + " ms");}
}

原有实现并不是线程安全的

class AccountUnsafe implements Account {private Integer balance;public AccountUnsafe(Integer balance) {this.balance = balance;}@Overridepublic Integer getBalance() {return balance;}@Overridepublic void withdraw(Integer amount) {balance -= amount;}
}

执行测试代码

public static void main(String[] args) {Account.demo(new AccountUnsafe(10000));}

某次的执行结果

330 cost: 306 ms

为什么不安全

withdraw 方法

public void withdraw(Integer amount) {balance -= amount;
}

原因是因为 会出现指令交错的情况，因为正常的逻辑，比如一个i–的操作会分为四步，1.先获取值、2.获取要减的数、3.相减、4.写回。正常来说，如果所有的都按照这个顺序来执行的话不可能出现线程安全的问题，但是实际上不是这样的，多线程的时候，或许可以保证有序性，但是没办法保证指令交错，所以导致 可能的顺序是 11234234，这样就会出现线程不安全的情况拉。

解决思路-锁

首先想到的是给 Account 对象加锁（但是太笨重了）

class AccountUnsafe implements Account {private Integer balance;public AccountUnsafe(Integer balance) {this.balance = balance;}@Overridepublic synchronized Integer getBalance() {return balance;}@Overridepublic synchronized void withdraw(Integer amount) {balance -= amount;}
}

结果为

0 cost: 399 ms 

解决思路-无锁

class AccountSafe implements Account {private AtomicInteger balance;public AccountSafe(Integer balance) {this.balance = new AtomicInteger(balance);}@Overridepublic Integer getBalance() {return balance.get();}@Overridepublic void withdraw(Integer amount) {while (true) {int prev = balance.get();// 获取余额的最新值int next = prev - amount;// 要修改的余额if (balance.compareAndSet(prev, next)) { // 真正修改，如果成功，结束循环，如果失败，继续循环break;}}// 可以简化为下面的方法// balance.addAndGet(-1 * amount);}
}

执行测试代码

public static void main(String[] args) {Account.demo(new AccountSafe(10000));
}

某次的执行结果

0 cost: 302 ms

是否真的无锁呢？

我们通过 Java 中的 AtomicInteger类中的 getAndIncrement()来看下 CAS 底层是怎么实现的。

	public final int getAndIncrement() {return unsafe.getAndAddInt(this, valueOffset, 1);}

可以看到它是调用的Unsafe类的getAndAddInt方法

public final int getAndAddInt(Object obj, long offset, int delta) {int value;do {value= this.getIntVolatile(obj, offset);} while(!this.compareAndSwapInt(obj, offset, value, value + delta));return v;
}

可以看到该方法内部是先获取到该对象的偏移量对应的值(value)，然后调用 compareAndSwapInt 方法通过对比来修改该值，如果这个值和value一样，说明此过程中间没有

人修改该数据，此时可以将该地址的值改为 value+delta, 返回true,结束循环。否则，说明有人修改该地址处的值，返回false，继续下一次循环。

那么是怎么保证 compareAndSwapInt(CAS)的原子性呢？这个就由操作系统底层来提供了。

其实 CAS 的底层是 lock cmpxchg 指令（X86 架构），在单核 CPU 和多核 CPU 下都能够保证【比较-交换】的原子性。

在多核状态下，某个核执行到带 lock 的指令时，CPU 会让总线锁住，当这个核把此指令执行完毕，再开启总线。这个过程中不会被线程的调度机制所打断，保证了多个线程对内存操作的准确性，是原子的。

CAS 与 volatile

前面看到的 AtomicInteger 的解决方法，内部并没有用锁来保护共享变量的线程安全。那么它是如何实现的呢？

public void withdraw(Integer amount) {// 需要不断尝试，直到成功为止while (true) {// 比如拿到了旧值 1000int prev = balance.get();// 在这个基础上 1000-10 = 990int next = prev - amount;/*compareAndSet 正是做这个检查，在 set 前，先比较 prev 与当前值- 不一致了，next 作废，返回 false 表示失败比如，别的线程已经做了减法，当前值已经被减成了 990那么本线程的这次 990 就作废了，进入 while 下次循环重试- 一致，以 next 设置为新值，返回 true 表示成功*/if (balance.compareAndSet(prev, next)) {break;}}}

其中的关键是 compareAndSet，它的简称就是 CAS （也有 Compare And Swap 的说法），它必须是原子操作。

其中，左侧的两个cas操作都失败了。

慢动作分析

@Slf4j
public class SlowMotion {public static void main(String[] args) {AtomicInteger balance = new AtomicInteger(10000);int mainPrev = balance.get();log.debug("try get {}", mainPrev);new Thread(() -> {sleep(1000);int prev = balance.get();balance.compareAndSet(prev, 9000);log.debug(balance.toString());}, "t1").start();sleep(2000);log.debug("try set 8000...");boolean isSuccess = balance.compareAndSet(mainPrev, 8000);log.debug("is success ? {}", isSuccess);if(!isSuccess){mainPrev = balance.get();log.debug("try set 8000...");isSuccess = balance.compareAndSet(mainPrev, 8000);log.debug("is success ? {}", isSuccess);}}private static void sleep(int millis) {try {Thread.sleep(millis);} catch (InterruptedException e) {e.printStackTrace();}}
}

输出结果

2023-10-13 11:28:37.134 [main] try get 10000 
2023-10-13 11:28:38.154 [t1] 9000 
2023-10-13 11:28:39.154 [main] try set 8000... 
2023-10-13 11:28:39.154 [main] is success ? false 
2023-10-13 11:28:39.154 [main] try set 8000... 
2023-10-13 11:28:39.154 [main] is success ? true 

volatile

AtomicInteger 源码里面 应用到了 volatile

获取共享变量时，为了保证该变量的可见性，需要使用 volatile 修饰。

它可以用来修饰成员变量和静态成员变量，他可以避免线程从自己的工作缓存中查找变量的值，必须到主存中获取
它的值，线程操作 volatile 变量都是直接操作主存。即一个线程对 volatile 变量的修改，对另一个线程可见。

注意：

volatile 仅仅保证了共享变量的可见性，让其它线程能够看到最新值，但不能解决指令交错问题（不能保证原
子性）

CAS 必须借助 volatile 才能读取到共享变量的最新值来实现【比较并交换】的效果

为什么无锁效率高

无锁情况下，即使重试失败，线程始终在高速运行，没有停歇，而 synchronized 会让线程在没有获得锁的时
候，发生上下文切换，进入阻塞。打个比喻

线程就好像高速跑道上的赛车，高速运行时，速度超快，一旦发生上下文切换，就好比赛车要减速、熄火，
等被唤醒又得重新打火、启动、加速… 恢复到高速运行，代价比较大。（cas 不会让线程停下来，while(true) 不停的循环）

但无锁情况下，因为线程要保持运行，需要额外 CPU 的支持，CPU 在这里就好比高速跑道，没有额外的跑
道，线程想高速运行也无从谈起，虽然不会进入阻塞，但由于没有分到时间片，仍然会进入可运行状态，还
是会导致上下文切换。（在多核cpu下能发挥出优势，虽然没有陷入block阻塞，但是没有分到时间片，还是要上下文切换）

CAS 的特点

结合 CAS 和 volatile 可以实现无锁并发，适用于线程数少、多核 CPU 的场景下。

• CAS 是基于乐观锁的思想：最乐观的估计，不怕别的线程来修改共享变量，就算改了也没关系，我吃亏点再
  重试呗。
• synchronized 是基于悲观锁的思想：最悲观的估计，得防着其它线程来修改共享变量，我上了锁你们都别想
  改，我改完了解开锁，你们才有机会。
• CAS 体现的是无锁并发、无阻塞并发，请仔细体会这两句话的意思，因为没有使用 synchronized，所以线程不会陷入阻塞，这是效率提升的因素之一、但如果竞争激烈，可以想到重试必然频繁发生，反而效率会受影响

原子整数

J.U.C 并发包提供了：

• AtomicBoolean
• AtomicInteger
• AtomicLong

以 AtomicInteger 为例

AtomicInteger i = new AtomicInteger(0);// 获取并自增（i = 0, 结果 i = 1, 返回 0），类似于 i++
System.out.println(i.getAndIncrement());// 自增并获取（i = 1, 结果 i = 2, 返回 2），类似于 ++i
System.out.println(i.incrementAndGet());// 自减并获取（i = 2, 结果 i = 1, 返回 1），类似于 --i
System.out.println(i.decrementAndGet());// 获取并自减（i = 1, 结果 i = 0, 返回 1），类似于 i--
System.out.println(i.getAndDecrement());// 获取并加值（i = 0, 结果 i = 5, 返回 0）
System.out.println(i.getAndAdd(5));// 加值并获取（i = 5, 结果 i = 0, 返回 0）
System.out.println(i.addAndGet(-5));

以getAndIncrement 源码为例

具体来说，这个方法会先读取对象var1上偏移量为var2的整数值，然后将其与给定的var4相加，在尝试使用CAS（Compare And Swap）操作将它们的和写回到这个偏移量上存储的值中。如果CAS操作成功，那么方法返回更新前的偏移量上存储的值，否则就重复执行这个过程，直到CAS操作成功为止。

// 获取并更新（i = 0, p 为 i 的当前值, 结果 i = -2, 返回 0）
// 其中函数中的操作能保证原子，但函数需要无副作用
System.out.println(i.getAndUpdate(p -> p - 2));// 更新并获取（i = -2, p 为 i 的当前值, 结果 i = 0, 返回 0）
// 其中函数中的操作能保证原子，但函数需要无副作用
System.out.println(i.updateAndGet(p -> p + 2));

原子引用

为什么需要原子引用类型？

• AtomicReference
• AtomicMarkableReference
• AtomicStampedReference

有如下方法

public interface DecimalAccount {// 获取余额BigDecimal getBalance();// 取款void withdraw(BigDecimal amount);/*** 方法内会启动 1000 个线程，每个线程做 -10 元 的操作* 如果初始余额为 10000 那么正确的结果应当是 0*/static void demo(DecimalAccount account) {List<Thread> ts = new ArrayList<>();for (int i = 0; i < 1000; i++) {ts.add(new Thread(() -> {account.withdraw(BigDecimal.TEN);}));}ts.forEach(Thread::start);ts.forEach(t -> {try {t.join();} catch (InterruptedException e) {e.printStackTrace();}});System.out.println(account.getBalance());}
}

试着提供不同的 DecimalAccount 实现，实现安全的取款操作

不安全实现

class DecimalAccountUnsafe implements DecimalAccount {BigDecimal balance;public DecimalAccountUnsafe(BigDecimal balance) {this.balance = balance;}@Overridepublic BigDecimal getBalance() {return balance;}@Overridepublic void withdraw(BigDecimal amount) {BigDecimal balance = this.getBalance();this.balance = balance.subtract(amount);}
}

安全实现-使用锁

class DecimalAccountSafeLock implements DecimalAccount {private final Object lock = new Object();BigDecimal balance;public DecimalAccountSafeLock(BigDecimal balance) {this.balance = balance;}@Overridepublic BigDecimal getBalance() {return balance;}@Overridepublic void withdraw(BigDecimal amount) {synchronized (lock) {BigDecimal balance = this.getBalance();this.balance = balance.subtract(amount);}}
}

安全实现-使用 CAS

class DecimalAccountSafeCas implements DecimalAccount {AtomicReference<BigDecimal> ref;public DecimalAccountSafeCas(BigDecimal balance) {ref = new AtomicReference<>(balance);}@Overridepublic BigDecimal getBalance() {return ref.get();}@Overridepublic void withdraw(BigDecimal amount) {while (true) {BigDecimal prev = ref.get();BigDecimal next = prev.subtract(amount);if (ref.compareAndSet(prev, next)) {break;}}}
}

测试代码

DecimalAccount.demo(new DecimalAccountUnsafe(new BigDecimal("10000")));
DecimalAccount.demo(new DecimalAccountSafeLock(new BigDecimal("10000")));
DecimalAccount.demo(new DecimalAccountSafeCas(new BigDecimal("10000")));

运行结果

4310 cost: 425 ms 
0 cost: 285 ms 
0 cost: 274 ms

ABA 问题及解决

ABA 问题

	static AtomicReference<String> ref = new AtomicReference<>("A");public static void main(String[] args) throws InterruptedException {log.debug("main start...");// 获取值 A// 这个共享变量被它线程修改过？String prev = ref.get();other();sleep(1);// 尝试改为 Clog.debug("change A->C {}", ref.compareAndSet(prev, "C"));}private static void other() {new Thread(() -> {log.debug("change A->B {}", ref.compareAndSet(ref.get(), "B"));}, "t1").start();sleep(0.5);new Thread(() -> {log.debug("change B->A {}", ref.compareAndSet(ref.get(), "A"));}, "t2").start();}

输出

11:29:52.325 c.Test36 [main] - main start... 
11:29:52.379 c.Test36 [t1] - change A->B true 
11:29:52.879 c.Test36 [t2] - change B->A true 
11:29:53.880 c.Test36 [main] - change A->C true 

主线程仅能判断出共享变量的值与最初值 A 是否相同，不能感知到这种从 A 改为 B 又 改回 A 的情况，如果主线程
希望：

只要有其它线程【动过了】共享变量，那么自己的 cas 就算失败，这时，仅比较值是不够的，需要再加一个版本号

AtomicStampedReference

	static AtomicStampedReference<String> ref = new AtomicStampedReference<>("A", 0);public static void main(String[] args) throws InterruptedException {log.debug("main start...");// 获取值 AString prev = ref.getReference();// 获取版本号int stamp = ref.getStamp();log.debug("版本 {}", stamp);// 如果中间有其它线程干扰，发生了 ABA 现象other();sleep(1);// 尝试改为 Clog.debug("change A->C {}", ref.compareAndSet(prev, "C", stamp, stamp + 1));}private static void other() {new Thread(() -> {log.debug("change A->B {}", ref.compareAndSet(ref.getReference(), "B",ref.getStamp(), ref.getStamp() + 1));log.debug("更新版本为 {}", ref.getStamp());}, "t1").start();sleep(0.5);new Thread(() -> {log.debug("change B->A {}", ref.compareAndSet(ref.getReference(), "A",ref.getStamp(), ref.getStamp() + 1));log.debug("更新版本为 {}", ref.getStamp());}, "t2").start();}

输出为

15:41:34.891 c.Test36 [main] - main start... 
15:41:34.894 c.Test36 [main] - 版本 0 
15:41:34.956 c.Test36 [t1] - change A->B true 
15:41:34.956 c.Test36 [t1] - 更新版本为 1 
15:41:35.457 c.Test36 [t2] - change B->A true 
15:41:35.457 c.Test36 [t2] - 更新版本为 2 
15:41:36.457 c.Test36 [main] - change A->C false 

AtomicStampedReference 可以给原子引用加上版本号，追踪原子引用整个的变化过程，如： A -> B -> A ->
C ，通过AtomicStampedReference，我们可以知道，引用变量中途被更改了几次。

但是有时候，并不关心引用变量更改了几次，只是单纯的关心是否更改过，所以就有了AtomicMarkableReference

AtomicMarkableReference

class GarbageBag {String desc;public GarbageBag(String desc) {this.desc = desc;}public void setDesc(String desc) {this.desc = desc;}@Overridepublic String toString() {return super.toString() + " " + desc;}
}

@Slf4j
public class TestABAAtomicMarkableReference {public static void main(String[] args) throws InterruptedException {GarbageBag bag = new GarbageBag("装满了垃圾");// 参数2 mark 可以看作一个标记，表示垃圾袋满了AtomicMarkableReference<GarbageBag> ref = new AtomicMarkableReference<>(bag, true);log.debug("主线程 start...");GarbageBag prev = ref.getReference();log.debug(prev.toString());new Thread(() -> {log.debug("打扫卫生的线程 start...");bag.setDesc("空垃圾袋");while (!ref.compareAndSet(bag, bag, true, false)) {} // 如果状态被改成了false，那么下次compareAndSet就不会成功了log.debug(bag.toString());}).start();Thread.sleep(1000);log.debug("主线程想换一只新垃圾袋？");boolean success = ref.compareAndSet(prev, new GarbageBag("空垃圾袋"), true, false);log.debug("换了么？" + success);log.debug(ref.getReference().toString());}
}

输出

2023-10-13 15:30:09.264 [main] 主线程 start... 
2023-10-13 15:30:09.270 [main] cn.itcast.GarbageBag@5f0fd5a0 装满了垃圾
2023-10-13 15:30:09.293 [Thread-1] 打扫卫生的线程 start... 
2023-10-13 15:30:09.294 [Thread-1] cn.itcast.GarbageBag@5f0fd5a0 空垃圾袋
2023-10-13 15:30:10.294 [main] 主线程想换一只新垃圾袋？
2023-10-13 15:30:10.294 [main] 换了么？false 
2023-10-13 15:30:10.294 [main] cn.itcast.GarbageBag@5f0fd5a0 空垃圾袋

原子数组

• AtomicIntegerArray
• AtomicLongArray
• AtomicReferenceArray

    /**参数1，提供数组、可以是线程不安全数组或线程安全数组参数2，获取数组长度的方法参数3，自增方法，回传 array, index参数4，打印数组的方法*/
// supplier 提供者 无中生有 ()->结果
// function 函数 一个参数一个结果 (参数)->结果 , BiFunction (参数1,参数2)->结果
// consumer 消费者 一个参数没结果 (参数)->void, BiConsumer (参数1,参数2)->private static <T> void demo(Supplier<T> arraySupplier,Function<T, Integer> lengthFun,BiConsumer<T, Integer> putConsumer,Consumer<T> printConsumer ) {List<Thread> ts = new ArrayList<>();T array = arraySupplier.get();int length = lengthFun.apply(array);for (int i = 0; i < length; i++) {// 每个线程对数组作 10000 次操作ts.add(new Thread(() -> {for (int j = 0; j < 10000; j++) {putConsumer.accept(array, j%length);}}));}ts.forEach(t -> t.start()); // 启动所有线程ts.forEach(t -> {try {t.join();} catch (InterruptedException e) {e.printStackTrace();}}); // 等所有线程结束printConsumer.accept(array);}

不安全的数组

demo(()->new int[10],(array)->array.length,(array, index) -> array[index]++,array-> System.out.println(Arrays.toString(array))
);

结果

[9870, 9862, 9774, 9697, 9683, 9678, 9679, 9668, 9680, 9698] 

其实本质上是这样的：
其执行的是，多个线程，在对应数组里面进行 ++操作，那么就会有这么一种情况，两个线程 刚好读到了同样的位置，然后 都对同一个数进行了 ++ 操作，此时，理论上 两个线程的++操作 最后加了2，实际上确加了1

安全的数组

demo(()-> new AtomicIntegerArray(10),(array) -> array.length(),(array, index) -> array.getAndIncrement(index),array -> System.out.println(array)
);

结果

[10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000] 

字段更新器

• AtomicReferenceFieldUpdater // 域 字段
• AtomicIntegerFieldUpdater
• AtomicLongFieldUpdater

利用字段更新器，可以针对对象的某个域（Field）进行原子操作，只能配合 volatile 修饰的字段使用，否则会出现异常

Exception in thread "main" java.lang.IllegalArgumentException: Must be volatile type

public class Test5 {private volatile int field;public static void main(String[] args) {AtomicIntegerFieldUpdater fieldUpdater =AtomicIntegerFieldUpdater.newUpdater(Test5.class, "field");Test5 test5 = new Test5();fieldUpdater.compareAndSet(test5, 0, 10);// 修改成功 field = 10System.out.println(test5.field);// 修改成功 field = 20fieldUpdater.compareAndSet(test5, 10, 20);System.out.println(test5.field);// 修改失败 field = 20fieldUpdater.compareAndSet(test5, 10, 30);System.out.println(test5.field);}
}

原子累加器（并发编程大师 - 与人类优秀的灵魂对话版）

剑指JUC原理-10.并发编程大师的原子累加器底层优化原理（与人类的优秀灵魂对话）-CSDN博客

Unsafe

cas 底层是调用的 unsafe

概述

Unsafe 对象提供了非常底层的，操作内存、线程的方法，Unsafe 对象不能直接调用，只能通过反射获得

public class UnsafeAccessor {static Unsafe unsafe;static {try {Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");theUnsafe.setAccessible(true);unsafe = (Unsafe) theUnsafe.get(null);} catch (NoSuchFieldException | IllegalAccessException e) {throw new Error(e);}}static Unsafe getUnsafe() {return unsafe;}
}

不要被名字所迷惑，名字虽然叫 Unsafe，但是这里并不是指什么线程安全的方面的不安全，而是指这个类比较底层，操作的都是内存，线程，不建议我们编程人员直接对它使用

Unsafe CAS 操作

其底层是通过内存偏移量来定位到这个属性，定位到属性以后，再对里面的属性值做一个比较并交换的动作。

@Data
class Student {volatile int id;volatile String name;
}

Unsafe unsafe = UnsafeAccessor.getUnsafe();
Field id = Student.class.getDeclaredField("id");
Field name = Student.class.getDeclaredField("name");// 获得成员变量的偏移量
long idOffset = UnsafeAccessor.unsafe.objectFieldOffset(id);
long nameOffset = UnsafeAccessor.unsafe.objectFieldOffset(name);
Student student = new Student();// 使用 cas 方法替换成员变量的值
UnsafeAccessor.unsafe.compareAndSwapInt(student, idOffset, 0, 20); // 返回 true
UnsafeAccessor.unsafe.compareAndSwapObject(student, nameOffset, null, "张三"); // 返回 true
System.out.println(student);

输出

Student(id=20, name=张三) 

使用自定义的 AtomicData 实现之前线程安全的原子整数 Account 实现

class AtomicData {private volatile int data;static final Unsafe unsafe;static final long DATA_OFFSET;static {unsafe = UnsafeAccessor.getUnsafe();try {// data 属性在 DataContainer 对象中的偏移量，用于 Unsafe 直接访问该属性DATA_OFFSET = unsafe.objectFieldOffset(AtomicData.class.getDeclaredField("data"));} catch (NoSuchFieldException e) {throw new Error(e);}}public AtomicData(int data) {this.data = data;}public void decrease(int amount) {int oldValue;while(true) {// 获取共享变量旧值，可以在这一行加入断点，修改 data 调试来加深理解oldValue = data;// cas 尝试修改 data 为 旧值 + amount，如果期间旧值被别的线程改了，返回 falseif (unsafe.compareAndSwapInt(this, DATA_OFFSET, oldValue, oldValue - amount)) {return;}}}public int getData() {return data;}
}

Account 实现

Account.demo(new Account() {AtomicData atomicData = new AtomicData(10000);@Overridepublic Integer getBalance() {return atomicData.getData();}@Overridepublic void withdraw(Integer amount) {atomicData.decrease(amount);}
});