CAS 详解
什么是 CAS？
CAS（Compare And Swap，即比较与交换）是非阻塞同步的底层实现原理，它是一种 CPU 硬件层面的指令，能够在 CPU 层面保证“比较并交换”这两个操作的原子性。
CAS 指令操作通常包含三个参数：
- 内存值（内存地址值）V
- 预期值 E
- 新值 N
当执行 CAS 指令时，当且仅当预期值 E 和内存值 V 相同时，才会将内存值更新为 N；否则不执行更新。无论是否更新成功，该操作都会返回旧的内存值 V。上述整个处理过程是一个不可分割的原子操作。
我们可以用一段 Java 伪代码来等效模拟一下 CAS 的执行过程：
public class CASDemo {
// 内存中当前的值
private volatile int ramAddress;
/**
* @param expectedValue 期望值
* @return newValue 更新的值
**/
public synchronized int compareAndSwap(int expectedValue, int newValue) {
// 模拟直接从内存地址读取到的当前值
int oldRamAddress = accessMemory(ramAddress);
// 将内存中的值与期望的值进行比较
if (oldRamAddress == expectedValue) {
ramAddress = newValue;
}
return oldRamAddress;
}
private int accessMemory(int ramAddress) {
// 模拟直接从内存地址读取到内存中的值
return ramAddress;
}
}以上伪代码描述了一个由“比较”和“赋值”两个阶段组成的复合操作。CAS 可以看作是将它们合并后的整体——一个不可分割的原子操作，并且其原子性是直接在底层硬件层面得到保障的。
CAS 是一种无锁算法，它在不使用传统锁机制（即不让线程阻塞）的情况下实现了多线程之间的变量同步。CAS 也是乐观锁（对比数据库中的悲观锁与乐观锁）的一种典型实现方式，Java 原子类中的自增操作就是通过 CAS 自旋来实现的。
CAS 的使用
在 Java 中，CAS 操作主要由 Unsafe 类提供支持，该类定义了三种针对不同类型变量的 CAS 操作：
/**
* Atomically updates Java variable to {@code x} if it is currently
* holding {@code expected}.
*
* <p>This operation has memory semantics of a {@code volatile} read
* and write. Corresponds to C11 atomic_compare_exchange_strong.
*
* @return {@code true} if successful
*/
@ForceInline
public final boolean compareAndSwapObject(Object o, long offset,
Object expected,
Object x) {
return theInternalUnsafe.compareAndSetReference(o, offset, expected, x);
}
/**
* Atomically updates Java variable to {@code x} if it is currently
* holding {@code expected}.
*
* <p>This operation has memory semantics of a {@code volatile} read
* and write. Corresponds to C11 atomic_compare_exchange_strong.
*
* @return {@code true} if successful
*/
@ForceInline
public final boolean compareAndSwapInt(Object o, long offset,
int expected,
int x) {
return theInternalUnsafe.compareAndSetInt(o, offset, expected, x);
}
/**
* Atomically updates Java variable to {@code x} if it is currently
* holding {@code expected}.
*
* <p>This operation has memory semantics of a {@code volatile} read
* and write. Corresponds to C11 atomic_compare_exchange_strong.
*
* @return {@code true} if successful
*/
@ForceInline
public final boolean compareAndSwapLong(Object o, long offset,
long expected,
long x) {
return theInternalUnsafe.compareAndSetLong(o, offset, expected, x);
}这三个方法调用的都是 native 方法，由 Java 虚拟机提供具体的底层实现。这意味着不同的 Java 虚拟机对它们的实现可能会略有不同。
Unsafe 是位于 sun.misc 包下的一个类，主要提供一些用于执行低级别、不安全操作的方法，如直接访问系统内存资源、自主管理内存等。这些方法在提升 Java 运行效率、增强底层资源操作能力方面起到了巨大作用。但同时，由于 Unsafe 类赋予了 Java 类似 C 语言指针那样直接操作内存空间的能力，这也增加了程序发生指针相关错误的风险。过度或不正确地使用 Unsafe 类会使程序出错的概率变大，让 Java 这种安全的语言变得不再“安全”，因此在业务开发中对 Unsafe 的使用一定要慎重。
以 compareAndSwapInt 为例，该方法接收 4 个参数：对象实例、内存偏移量、字段期望值、字段新值。该方法会针对指定对象实例中对应偏移量的字段执行 CAS 操作。
import lombok.Data;
import sun.misc.Unsafe;
import java.lang.reflect.Field;
public class CASTest {
public static void main(String[] args) {
Entity entity = new Entity();
Unsafe unsafe = UnsafeFactory.getUnsafe();
long offset = UnsafeFactory.getFieldOffset(unsafe, Entity.class, "x");
boolean successful;
// 4个参数分别是：对象实例、字段的内存偏移量、字段期望值、字段新值
successful = unsafe.compareAndSwapInt(entity, offset, 0, 3);
System.out.println(successful + "\t" + entity.x);
successful = unsafe.compareAndSwapInt(entity, offset, 3, 5);
System.out.println(successful + "\t" + entity.x);
successful = unsafe.compareAndSwapInt(entity, offset, 3, 8);
System.out.println(successful + "\t" + entity.x);
}
static class UnsafeFactory {
/**
* 反射获取 Unsafe 对象
*
* @return Unsafe 实例
*/
public static Unsafe getUnsafe() {
try {
Field field = Unsafe.class.getDeclaredField("theUnsafe");
field.setAccessible(true);
return (Unsafe) field.get(null);
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
/**
* 获取字段的内存偏移量
*/
public static long getFieldOffset(Unsafe unsafe, Class clazz, String fieldName) {
try {
return unsafe.objectFieldOffset(clazz.getDeclaredField(fieldName));
} catch (NoSuchFieldException e) {
throw new Error(e);
}
}
}
@Data
static class Entity {
private int x = 0;
}
}针对 entity.x 的 3 次 CAS 操作，执行结果如下：
true 3
true 5
false 5CAS 的应用场景
CAS 在 java.util.concurrent.atomic 包下的相关类、Java AQS、以及 ConcurrentHashMap 等底层实现中有着极其广泛的应用。
如下图所示，在 AtomicInteger 的实现中，静态字段 valueOffset 即为内部属性 value 的内存偏移地址。该值在 AtomicInteger 类的静态代码块中通过 Unsafe.objectFieldOffset 方法初始化获取。
当调用 AtomicInteger 提供的线程安全方法时，底层会通过 valueOffset 定位到当前对象中 value 的真实内存地址，从而利用 CAS 实现对 value 字段的原子操作。
private static final Unsafe U = Unsafe.getUnsafe();
private static final long VALUE
= U.objectFieldOffset(AtomicInteger.class, "value");
private volatile int value;AtomicInteger 对象在执行自增操作，假设对象的基地址 baseAddress = "0x110000"，通过 baseAddress + valueOffset 即可计算出 value 的内存地址 valueAddress = "0x11000c"。接着通过 CAS 进行原子性的更新操作，如果成功则返回，如果失败则继续重试（自旋），直到更新成功为止。
CAS 的缺陷
虽然 CAS 高效地解决了并发环境下的原子操作问题，但它依然存在一些明显的缺陷：
- 自旋开销大：如果并发量大且 CAS 长时间未能成功，自旋重试会给 CPU 带来极大的开销。
- 只能保证一个共享变量的原子操作：无法像锁一样锁定多个变量的复合操作。
- ABA 问题。
ABA 问题及其解决方案
CAS 算法实现的一个重要前提是：需要先取出内存中某个时刻的数据，然后在下一时刻进行比较并替换。在这个时间差内，数据有可能已经发生了我们无法察觉的变化。
什么是 ABA 问题
当有多个线程对同一个原子类进行操作时，假设线程1刚读取了值为 A，此时线程2迅速将该值修改为 B，然后紧接着又修改回了 A。当线程1再次去进行 CAS 比较时，发现当前值依然是 A，于是成功执行了更新。线程1完全感知不到中间曾有其他线程介入修改过。
代码演示：
@Slf4j
public class ABATest {
public static void main(String[] args) {
AtomicInteger atomicInteger = new AtomicInteger(1);
new Thread(()->{
int value = atomicInteger.get();
log.debug("Thread1 read value: " + value);
// 刻意阻塞1s，给线程2制造修改机会
LockSupport.parkNanos(1000000000L);
// Thread1尝试通过CAS将value值修改为3
if (atomicInteger.compareAndSet(value, 3)) {
log.debug("Thread1 update from " + value + " to 3");
} else {
log.debug("Thread1 update fail!");
}
},"Thread1").start();
new Thread(()->{
int value = atomicInteger.get();
log.debug("Thread2 read value: " + value);
// Thread2先通过CAS将value值从1修改为2
if (atomicInteger.compareAndSet(value, 2)) {
log.debug("Thread2 update from " + value + " to 2");
// 模拟执行一些其他业务
value = atomicInteger.get();
log.debug("Thread2 read value: " + value);
// Thread2紧接着再通过CAS将value值从2修改回1 (此时形成了ABA)
if (atomicInteger.compareAndSet(value, 1)) {
log.debug("Thread2 update from " + value + " to 1");
}
}
},"Thread2").start();
}
}运行后发现，Thread1并不清楚Thread2对 value 进行了“狸猫换太子”的操作，误以为 value 从未被修改过，最终依然将值成功改为了3。
17:10:47.307 [Thread1] DEBUG com.juzicoding.cas.ABATest -- Thread1 read value: 1
17:10:47.307 [Thread2] DEBUG com.juzicoding.cas.ABATest -- Thread2 read value: 1
17:10:47.308 [Thread2] DEBUG com.juzicoding.cas.ABATest -- Thread2 update from 1 to 2
17:10:47.308 [Thread2] DEBUG com.juzicoding.cas.ABATest -- Thread2 read value: 2
17:10:47.308 [Thread2] DEBUG com.juzicoding.cas.ABATest -- Thread2 update from 2 to 1
17:10:48.309 [Thread1] DEBUG com.juzicoding.cas.ABATest -- Thread1 update from 1 to 3ABA 问题的解决方案
在数据库中，我们常用一种基于“数据版本号”的乐观锁机制来实现数据同步：每次修改数据时，版本号就会递增。
Java 也借鉴了这个思想，提供了相应的原子引用类 AtomicStampedReference<V>。
public class AtomicStampedReference<V> {
private static class Pair<T> {
final T reference;
final int stamp;
private Pair(T reference, int stamp) {
this.reference = reference;
this.stamp = stamp;
}
static <T> Pair<T> of(T reference, int stamp) {
return new Pair<T>(reference, stamp);
}
}
private volatile Pair<V> pair;
}其中，reference 表示我们实际要存储的变量，stamp 则是版本戳（或标记）。每次进行修改时，除了要校验期望值，还要校验版本号，并让版本号+1。这保证了即使值回退到了原状态，版本号依然是向前递增的，从而彻底打破了ABA困境。
@Slf4j
public class AtomicStampedReferenceTest {
public static void main(String[] args) {
// 定义AtomicStampedReference，Pair.reference初始值为1, Pair.stamp版本号为1
AtomicStampedReference atomicStampedReference = new AtomicStampedReference(1,1);
new Thread(()->{
int[] stampHolder = new int[1];
int value = (int) atomicStampedReference.get(stampHolder);
int stamp = stampHolder[0];
log.debug("Thread1 read value: " + value + ", stamp: " + stamp);
// 阻塞1s
LockSupport.parkNanos(1000000000L);
// Thread1尝试通过CAS修改value值为3，同时期望stamp为之前的stamp，并将stamp+1
if (atomicStampedReference.compareAndSet(value, 3, stamp, stamp+1)) {
log.debug("Thread1 update from " + value + " to 3");
} else {
log.debug("Thread1 update fail!");
}
},"Thread1").start();
new Thread(()->{
int[] stampHolder = new int[1];
int value = (int)atomicStampedReference.get(stampHolder);
int stamp = stampHolder[0];
log.debug("Thread2 read value: " + value+ ", stamp: " + stamp);
// Thread2先修改为2
if (atomicStampedReference.compareAndSet(value, 2, stamp, stamp+1)) {
log.debug("Thread2 update from " + value + " to 2");
value = (int) atomicStampedReference.get(stampHolder);
stamp = stampHolder[0];
log.debug("Thread2 read value: " + value+ ", stamp: " + stamp);
// Thread2再修改回1
if (atomicStampedReference.compareAndSet(value, 1, stamp, stamp+1)) {
log.debug("Thread2 update from " + value + " to 1");
}
}
},"Thread2").start();
}
}通过引入版本号，这次 Thread1 感知到了中间发生的变化，更新失败。
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.atomic.AtomicStampedReference;
import java.util.concurrent.locks.LockSupport;
@Slf4j
public class AtomicStampedReferenceTest {
public static void main(String[] args) {
// 定义AtomicStampedReference，Pair.reference初始值为1, Pair.stamp版本号为1
AtomicStampedReference atomicStampedReference = new AtomicStampedReference(1, 1);
new Thread(() -> {
int[] stampHolder = new int[1];
int value = (int) atomicStampedReference.get(stampHolder);
int stamp = stampHolder[0];
log.debug("Thread1 read value: " + value + ", stamp: " + stamp);
// 阻塞1s
LockSupport.parkNanos(1000000000L);
// Thread1尝试通过CAS修改value值为3，同时期望stamp为之前的stamp，并将stamp+1
if (atomicStampedReference.compareAndSet(value, 3, stamp, stamp + 1)) {
log.debug("Thread1 update from " + value + " to 3");
} else {
log.debug("Thread1 update fail!");
}
}, "Thread1").start();
new Thread(() -> {
int[] stampHolder = new int[1];
int value = (int) atomicStampedReference.get(stampHolder);
int stamp = stampHolder[0];
log.debug("Thread2 read value: " + value + ", stamp: " + stamp);
// Thread2先修改为2
if (atomicStampedReference.compareAndSet(value, 2, stamp, stamp + 1)) {
log.debug("Thread2 update from " + value + " to 2");
value = (int) atomicStampedReference.get(stampHolder);
stamp = stampHolder[0];
log.debug("Thread2 read value: " + value + ", stamp: " + stamp);
// Thread2再修改回1
if (atomicStampedReference.compareAndSet(value, 1, stamp, stamp + 1)) {
log.debug("Thread2 update from " + value + " to 1");
}
}
}, "Thread2").start();
}
}另外 Java 还提供了AtomicMarkableReference，AtomicMarkableReference 可以理解为 AtomicStampedReference 的简化版。它不关心数据被修改过了几次，仅仅关心数据是否被修改过。因此它的标记不再是整型的版本戳，而是一个 boolean 类型的 mark 变量。
public class AtomicMarkableReference<V> {
private static class Pair<T> {
final T reference;
final boolean mark;
private Pair(T reference, boolean mark) {
this.reference = reference;
this.mark = mark;
}
static <T> Pair<T> of(T reference, boolean mark) {
return new Pair<T>(reference, mark);
}
}
private volatile Pair<V> pair;
}