全方位探究似懂非懂的 CAS 机制～

Love The Way You Lie 2023-09-24 23:27 13阅读 0赞

### 前言 ###

> 求学、面试的时候会无法回避 CAS 话题，但对于其原理，总有种似懂非懂的感觉。

CAS 机制全称： *Compare and Swap*，即 **比较并替换，** 。也有叫做 *Compare and Set* 的，即**比较并设置**。顾名思义，分为两步：

1.  比较：读取到了一个值 A，在将其更新为 B 之前，检查原值是否仍为 A

1.  替换 / 设置：YES 则将 A 更新为 B，结束；反之，重复上述操作直到成功为止

这种机制在确保原子化操作、实现乐观锁的同时也无法避免一些缺陷，咱们从源码入手分析一下其原理、乐观锁和缺陷等各个细节。

### 源码解读 ###

以 JDK 中最常用的 AtomicInteger 类的源码为例进行探讨，版本为 **JDK 8**。

#### 1. Unsafe ####

首先 AtomicInteger 将通过 Unsafe 提供的静态方法 getUnsafe() 获得其单例。

// java.util.concurrent.atomic.AtomicInteger
     publicclassAtomicIntegerextendsNumberimplementsjava.io.Serializable {
         privatestaticfinalUnsafeunsafe= Unsafe.getUnsafe();
         ...
     }
     
     // jdk.internal.misc.Unsafe
     publicfinalclassUnsafe {
         ...
         privateUnsafe() {}
     
         privatestaticfinalUnsafetheUnsafe=newUnsafe();
     
         @CallerSensitive
         publicstatic Unsafe getUnsafe() {
             ...
             return theUnsafe;
         }
     }
    复制代码

> **为什么叫 Unsafe 呢？**  
> 因为它提供了针对任意地址的数据进行不安全读写等内存操作、线程调度、CAS 等操作的入口，如果调用是不受信任的，那么使得代码出错的概率变大，也更会引发 JVM 级别的 exception。  
> 所以要求只有 **可信任**的代码可以获取其单例并进行调用，所以将其命名为 Unsafe，给调用者以提醒。

正因为此，getUnsafe() 内部在返回 Unsafe 单例前会先去检查调用 Class 其是否可信任，如果不是系统 ClassLoader 加载的 Class 的话，会抛出 SecurityException。

// jdk.internal.misc.Unsafe
     public final classUnsafe {
         ...
         @CallerSensitive
         publicstaticUnsafegetUnsafe() {
             Class<?> caller = Reflection.getCallerClass();
             if (!VM.isSystemDomainLoader(caller.getClassLoader()))
                 thrownewSecurityException("Unsafe");
             return theUnsafe;
         }
     }
     
     // sun.misc.VM
     publicclassVM {
         ...
         publicstaticbooleanisSystemDomainLoader(ClassLoader loader) {
             return loader == null;
         }
     }
    复制代码

并建议像如下示例代码一样进行使用：

classMyTrustedClass {
         privatestatic final Unsafe unsafe = Unsafe.getUnsafe();
         ...
         privatelong myCountAddress = ...;
         publicintgetCount() { returnunsafe.getByte(myCountAddress); }
     }
    复制代码

#### 2. valueOffset & value ####

其次，赋值内部持有相关变量，包括目标的 int 型数值 *value* 和关联该 value 地址的 long 型的 *valueOffset*：

*  *value* 使用 volatile 修饰，默认为 0，反之为 AtomicInteger 构造时指定的初始值 *initialValue*

*  *valueOffset* 是存放 *value* 的内存相对地址：在 static 块中通过 Unsafe 的 objectFieldOffset() 传入的 *value* 的 Field 字段得到

// java.util.concurrent.atomic.AtomicInteger
     publicclassAtomicIntegerextendsNumberimplementsjava.io.Serializable {
         ...
         privatevolatileint value;
     
         publicAtomicInteger(int initialValue) {
             value = initialValue;
         }
     
         publicAtomicInteger() { }
     
         privatestaticfinallong valueOffset;
     
         static {
             try {
                 valueOffset = unsafe.objectFieldOffset
                     (AtomicInteger.class.getDeclaredField("value"));
             } catch (Exception ex) { thrownewError(ex); }
         }
     }
     
     // jdk.internal.misc.Unsafe
     publicfinalclassUnsafe {
         ...
         publicnativelongobjectFieldOffset(Field f);
     }
    复制代码

Unsafe\_ObjectFieldOffset 的实现是调用 find\_field\_offset 进行，其将进行 Field 的非空检查，以及 Klass 的转化。最后还要转换成 InstanceKlass 实例，并获取其中的 field\_offset 并返回。

// unsafe.cpp
     UNSAFE_ENTRY(jlong, Unsafe_ObjectFieldOffset(JNIEnv *env, jobject unsafe, jobject field))
       UnsafeWrapper("Unsafe_ObjectFieldOffset");
       return find_field_offset(field, 0, THREAD);
     UNSAFE_END
         
     jint find_field_offset(jobject field, int must_be_static, TRAPS) {
       if (field == NULL) {
         THROW_0(vmSymbols::java_lang_NullPointerException());
       }
     
       oop reflected   = JNIHandles::resolve_non_null(field);
       oop mirror      = java_lang_reflect_Field::clazz(reflected);
       Klass* k      = java_lang_Class::as_Klass(mirror);
       int slot        = java_lang_reflect_Field::slot(reflected);
       int modifiers   = java_lang_reflect_Field::modifiers(reflected);
     
       if (must_be_static >= 0) {
         int really_is_static = ((modifiers & JVM_ACC_STATIC) != 0);
         if (must_be_static != really_is_static) {
           THROW_0(vmSymbols::java_lang_IllegalArgumentException());
         }
       }
     
       int offset = InstanceKlass::cast(k)->field_offset(slot);
       return field_offset_from_byte_offset(offset);
     }
     
     // instanceKlass.h
     class InstanceKlass: public Klass {
       ...
      public:
       int     field_offset      (int index) const { return field(index)->offset(); }
     
         // Casting from Klass*
       static InstanceKlass* cast(Klass* k) {
         assert(k == NULL || k->is_klass(), "must be");
         assert(k == NULL || k->oop_is_instance(), "cast to InstanceKlass");
         return (InstanceKlass*) k;
       }
     }
    复制代码

#### 3. compareAndSwap ####

后面是在写值的时候继续调用 Unsafe，关键在于其 native 侧的实现。

比如 compareAndSet() 将传递实例自身，value 的内存相对地址，当前的期待值和目标的更新值四者传递给 Unsafe：

*  返回 false 表示当前的值并非期待值，无法完成更新

*  返回 true 表示更新成功

// java.util.concurrent.atomic.AtomicInteger
     publicclassAtomicIntegerextendsNumberimplementsjava.io.Serializable {
         privatestaticfinallong valueOffset;
         ...
     
         publicfinalbooleancompareAndSet(int expect, int update) {
             return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
         }
     }
     
     // jdk.internal.misc.Unsafe
     publicfinalclassUnsafe {
         ...
         publicfinalnativebooleancompareAndSwapInt(Object o, long offset,
                                                       int expected,
                                                       int x);
     }
    复制代码

native 的实现是 compareAndSwapInt()。

*  先通过 JNIHandles 找到 AtomicInteger 实例的地址 *oop*

*  将该 *oop* 和 *value* 的相对地址作为参数调用 index\_oop\_from\_field\_offset\_long() 并得到 value 的内存绝对地址 *aadr*

*  将更新值、*addr* 和期待值作为参数调用 cmpxchg() 继续

// unsafe.cpp
     UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x))
       UnsafeWrapper("Unsafe_CompareAndSwapInt");
       oop p = JNIHandles::resolve(obj);
       jint* addr = (jint *) index_oop_from_field_offset_long(p, offset);
       return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
     UNSAFE_END
         
     inline void* index_oop_from_field_offset_long(oop p, jlong field_offset) {
       jlong byte_offset = field_offset_to_byte_offset(field_offset);
     #ifdef ASSERT
       if (p != NULL) {
         assert(byte_offset >= 0 && byte_offset <= (jlong)MAX_OBJECT_SIZE, "sane offset");
         if (byte_offset == (jint)byte_offset) {
           void* ptr_plus_disp = (address)p + byte_offset;
           assert((void*)p->obj_field_addr<oop>((jint)byte_offset) == ptr_plus_disp,
                  "raw [ptr+disp] must be consistent with oop::field_base");
         }
         jlong p_size = HeapWordSize * (jlong)(p->size());
         assert(byte_offset < p_size, err_msg("Unsafe access: offset " INT64_FORMAT " > object's size " INT64_FORMAT, byte_offset, p_size));
       }
     #endif
       if (sizeof(char*) == sizeof(jint))    // (this constant folds!)
         return (address)p + (jint) byte_offset;
       else
         return (address)p +        byte_offset;
     }
    复制代码

cmpxchg() 的实现在 atomic.cpp 中，其中 value 的地址用 volatile 关键字描述，意味着目标数据的变化将立即反映到内存、并要求读取时应当从内存中取出。

1.  首先计算得到 value 地址的当前数值，即 cur\_as\_bytes\[offset\]

1.  假使当前数值等于期待值 compare\_value：

*  调用 cmpxchg CPU 指令将 dest\_int 地址对应的数值执行期待值为 cur，更新值为 new\_val 的 CAS 操作

*  该操作返回的是期待值的话，表示更新成功，跳过循环并返回期待值。Unsafe\_CompareAndSwapInt() 将收到匹配期待值的结果，并向 Java 侧返回 true

*  反之，将 cur 更新为 CPU 指令返回的当前值，再继续下一次循环，直到成功为止

1.  反之即不等于期待值，直接返回

*  Unsafe\_CompareAndSwapInt() 将收到不匹配期待值的结果，并向 Java 侧返回 false

// atomic.cpp
     jbyte Atomic::cmpxchg(jbyte exchange_value, volatile jbyte* dest, jbyte compare_value) {
       uintptr_t dest_addr = (uintptr_t)dest;
       uintptr_t offset = dest_addr % sizeof(jint);
         
       volatile jint* dest_int = (volatile jint*)(dest_addr - offset);
       jint cur = *dest_int;
         
       jbyte* cur_as_bytes = (jbyte*)(&cur);
       jint new_val = cur;
         
       jbyte* new_val_as_bytes = (jbyte*)(&new_val);
       new_val_as_bytes[offset] = exchange_value;
         
       while (cur_as_bytes[offset] == compare_value) {
         jint res = cmpxchg(new_val, dest_int, cur);
         if (res == cur) break;
         cur = res;
         new_val = cur;
         new_val_as_bytes[offset] = exchange_value;
       }
       return cur_as_bytes[offset];
     }
    复制代码

事实上，Android 中 AtomicInteger 的实现稍稍不同，没有只用 Unsafe 而是采用了 VarHandle，这里不再展开。

// android/libcore/ojluni/src/main/java/java/
     // util/concurrent/atomic/AtomicInteger.java
     public classAtomicIntegerextendsNumberimplementsjava.io.Serializable{
         ...
         public final boolean compareAndSet(int expectedValue, int newValue) {
              // Android-changed: Using VarHandle instead of Unsafe
              // return U.compareAndSetInt(this, VALUE, expectedValue, newValue);
              returnVALUE.compareAndSet(this, expectedValue, newValue);
          }
     }
    复制代码

### 乐观锁 ###

利用 CAS 机制可以实现一个**乐观锁**，即无需加锁可实现多个线程同时读取、但是仅有一个线程可以成功写入数据，并导致其他要卸乳数据的线程回滚重试。

例如 Unsafe 通过上述的 compareAndSwapInt() 实现的自增 1 的原子操作的逻辑。

// java.util.concurrent.atomic.AtomicInteger
     publicclassAtomicIntegerextendsNumberimplementsjava.io.Serializable {
         ...
         publicfinalintgetAndIncrement() {
             return unsafe.getAndAddInt(this, valueOffset, 1);
         }
     }
     
     // jdk.internal.misc.Unsafe
     publicfinalclassUnsafe {
         ...
         publicfinalintgetAndAddInt(Object o, long offset, int delta) {
             int v;
             do {
                 v = getIntVolatile(o, offset);
             } while (!compareAndSwapInt(o, offset, v, v + delta));
             return v;
         }
     }
    复制代码

因为整个过程中不涉及加/解锁的操作，乐观锁也被称为**无锁编程**。

> 本质上说，乐观锁其实不是“锁”，它仅仅是一个循环重试 CAS 的算法而已！

### 缺陷 ###

CAS 机制可以高效地实现原子操作，但仍不完美：

1.  循环时间长开销大：CAS 大量失败后长时间占用 CPU 资源，加大了系统性能开销

1.  只能保证一个共享变量的原子操作：当对一个共享变量执行操作时，我们可以使用循环 CAS 的方式来保证原子操作，但是对多个共享变量操作时，循环 CAS 就无法保证操作的原子性

1.  ABA 问题：CAS 机制本质上依赖值有没有发生变化的操作条件。但是如果值原来是 A、被改成变成了 B、最后又变回了 A，那么使用 CAS 进行检查时会发现它的值没有发生变化进行了操作，但是实际上却变化了，这其实违背了约定的条件。

> Java 1.5 开始提供了一个类AtomicStampedReference 来解决该问题，其提供的 compareAndSet() 会首先检查当前引用是否等于预期引用，并且当前标志是否等于预期标志，如果全部相等，才进行值的更新操作

### 结语 ###

可以看到无论是初始化还是写值都是通过 Unsafe 调度完成，整理了如下的流程图以加深理解：

![format_png][]

1.  AtomicInteger 通过静态方法 getUnsafe() 获得 Unsafe 实例

1.  接着通过 Unsafe 实例的 native 方法 objectFieldOffset() 传入使用 volatie 修饰的数值 *value* 的 Field 字段得到其内存相对地址

1.  关键的 compareAndSet() 亦由 Unsafe 调度，即 native 方法 compareAndSwapInt()。主要是传入 AtomicInteger 实例，value 内存相对地址，期待值以及更新值。

native 的实现由 atomic.cpp 的 cmpxchg() 完成，当前 value 内存地址返回的值匹配期待值的话，执行更新操作；反之，直接返回 false。

执行更新的操作采用 cmpxchg CPU 指令，如果得到的值不符合期待值的话，更新期待值继续下一次循环，直到匹配为止。

作者：TechMerger

链接：[https://juejin.cn/post/7209578803159924796][https_juejin.cn_post_7209578803159924796]

[format_png]: https://img-blog.csdnimg.cn/img_convert/1a630d186762b1940224f1a8de0d170c.webp?x-oss-process=image/format,png
[https_juejin.cn_post_7209578803159924796]: https://juejin.cn/post/7209578803159924796