并發編程利器 Java CAS 原子類全解

作者：碼農SharkChili 2024-11-21 14:55:37

本文將全面解析 Java 中的 CAS 原子類，探討其背后的原理、應用場景以及如何有效利用這些工具來提升程序的并發性能和安全性。

在現代軟件開發中，多線程編程已經成為構建高性能、高響應性應用程序的關鍵技術之一。然而，多線程環境下的數據一致性問題一直是開發者面臨的一大挑戰。為了解決這一難題，Java 平臺提供了多種并發控制工具和機制，其中 java.util.concurrent.atomic 包中的 CAS（Compare-and-Swap）原子類尤為突出。

本文將全面解析 Java 中的 CAS 原子類，探討其背后的原理、應用場景以及如何有效利用這些工具來提升程序的并發性能和安全性。通過深入理解 CAS 機制，讀者將能夠更好地應對復雜的并發編程場景，編寫出更加健壯和高效的代碼。

一、什么是CAS

CAS全稱Compare-And-Swap，是一種無鎖編程算法，即比較當前的值與舊值是否相等若相等則進行修改操作(樂觀鎖機制)，該類常用于多線程共享變量的修改操作。而其底層實現也是基于硬件平臺的匯編指令，JVM只是封裝其調用僅此而已。而本文會基于以下大綱展開對CAS的探討。

二、CAS基礎使用示例

如下所示，可以看出使用封裝CAS操作的AtomicInteger操作多線程共享變量無需我們手動加鎖，因為避免過多人為操作這就大大減少了多線程操作下的失誤。

使用原子類操作共享數據：

public class CasTest {
    private AtomicInteger count = new AtomicInteger();


    public void increment() {
        count.incrementAndGet();
    }
    // 使用 AtomicInteger 后，不需要加鎖，也可以實現線程安全
    public int getCount() {
        return count.get();
    }

    public static void main(String[] args) {

    }
}

使用sync鎖操作數據：

public class Test {
    private int i=0;
    public synchronized int add(){
        return i++;
    }
}

三、從源碼角度了解java如何封裝匯編的UNSAFE

代碼也很簡單，就是拿到具有可見性的volatile變量i，然后判斷i和當前對象paramObject對應的i值是否一致，若一致則說明沒被人該過，進而進行修改操作，反之自旋循環獲取在進行CAS。

public final int getAndAddInt(Object paramObject, long paramLong, int paramInt)
  {
    int i;
    do
      i = getIntVolatile(paramObject, paramLong);
    while (!compareAndSwapInt(paramObject, paramLong, i, i + paramInt));
    return i;
  }

  public final long getAndAddLong(Object paramObject, long paramLong1, long paramLong2)
  {
    long l;
    do
      l = getLongVolatile(paramObject, paramLong1);
    while (!compareAndSwapLong(paramObject, paramLong1, l, l + paramLong2));
    return l;
  }

  public final int getAndSetInt(Object paramObject, long paramLong, int paramInt)
  {
    int i;
    do
      i = getIntVolatile(paramObject, paramLong);
    while (!compareAndSwapInt(paramObject, paramLong, i, paramInt));
    return i;
  }

  public final long getAndSetLong(Object paramObject, long paramLong1, long paramLong2)
  {
    long l;
    do
      l = getLongVolatile(paramObject, paramLong1);
    while (!compareAndSwapLong(paramObject, paramLong1, l, paramLong2));
    return l;
  }

  public final Object getAndSetObject(Object paramObject1, long paramLong, Object paramObject2)
  {
    Object localObject;
    do
      localObject = getObjectVolatile(paramObject1, paramLong);
    while (!compareAndSwapObject(paramObject1, paramLong, localObject, paramObject2));
    return localObject;
  }

四、手寫Unsafe實現20個線程500次CAS自增

代碼邏輯和注釋如下，讀者可自行debug查看邏輯：

public class CasCountInc {

    private static Logger logger = LoggerFactory.getLogger(CasCountInc.class);

    // 獲取Unsafe對象
    private static Unsafe unsafe = getUnsafe();

    // 線程池數目
    private static final int THREAD_COUNT = 20;

    // 每個線程運行自增次數
    private static final int EVERY_THREAD_ADD_COUNT = 500;

    // 自增的count的值，volatile保證可見性
    private volatile int count = 0;

    // count字段的偏移量
    private static long countOffSet;

    private static Unsafe getUnsafe() {
        Unsafe unsafe = null;
        try {
            Field field = Unsafe.class.getDeclaredField("theUnsafe");
            field.setAccessible(true);
            unsafe = (Unsafe) field.get(null);
        } catch (Exception e) {
            logger.info("獲取unsafe失敗，失敗原因:[{}]", e.getMessage(), e);
        }
        return unsafe;
    }


    static {
        try {
            countOffSet = unsafe.objectFieldOffset(CasCountInc.class.getDeclaredField("count"));
        } catch (NoSuchFieldException e) {
            logger.error("獲取count的偏移量報錯，錯誤原因:[{}]", e.getMessage(), e);
        }
    }


    public void inc() {
        int oldCount = 0;
        //基于cas完成自增
        do {
            oldCount = count;
        } while (!unsafe.compareAndSwapInt(this, countOffSet, oldCount, oldCount + 1));
    }


    public static void main(String[] args) throws InterruptedException {
        CasCountInc casCountInc = new CasCountInc();
        CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT);
        IntStream.range(0, THREAD_COUNT).forEach(i -> {
            new Thread(() -> {
                IntStream.range(0, EVERY_THREAD_ADD_COUNT).forEach((j) -> {
                    casCountInc.inc();
                });
                countDownLatch.countDown();
            }).start();
        });

        countDownLatch.await();

        logger.info("count最終結果為 [{}]", casCountInc.count);
    }
}

五、原子類簡介

1.原子類更新基本類型

原子類基本類型的格式為Atomic+包裝類名，這里筆者列舉幾個比較常用的：

AtomicBoolean: 原子更新布爾類型。
AtomicInteger: 原子更新整型。
AtomicLong: 原子更新長整型。

2.原子類更新數組類型

AtomicIntegerArray: 原子更新整型數組里的元素。
AtomicLongArray: 原子更新長整型數組里的元素。
AtomicReferenceArray: 原子更新引用類型數組里的元素。

對應我們給出AtomicIntegerArray原子操作數組的示例：

public class AtomicIntegerArrayDemo {


    public static void main(String[] args) throws InterruptedException {
        AtomicIntegerArray array = new AtomicIntegerArray(new int[] { 0, 0 });
        System.out.println(array);
//        索引1位置+2
        System.out.println(array.getAndAdd(1, 2));
        System.out.println(array);
    }
}

3.原子類更新引用類型

AtomicReference: 原子更新引用類型。
AtomicStampedReference: 原子更新引用類型, 內部使用Pair來存儲元素值及其版本號。
AtomicMarkableReferce: 原子更新帶有標記位的引用類型。

對應的我們給出原子操作引用類型的代碼示例：

import java.util.concurrent.atomic.AtomicReference;

public class AtomicReferenceTest {

    public static void main(String[] args){

        // 創建兩個Person對象，它們的id分別是101和102。
        Person p1 = new Person(101);
        Person p2 = new Person(102);
        // 新建AtomicReference對象，初始化它的值為p1對象
        AtomicReference ar = new AtomicReference(p1);
        // 通過CAS設置ar。如果ar的值為p1的話，則將其設置為p2。
        ar.compareAndSet(p1, p2);

        Person p3 = (Person)ar.get();
        System.out.println("p3 is "+p3);
        System.out.println("p3.equals(p1)="+p3.equals(p1));
        System.out.println("p3.equals(p2)="+p3.equals(p2));
    }
}



class Person {
    volatile long id;
    public Person(long id) {
        this.id = id;
    }
    public String toString() {
        return "id:"+id;
    }
}

六、原子類更新成員變量

通過原子類型操作成員變量大體有以下幾個更新器：

AtomicIntegerFieldUpdater: 原子更新整型的字段的更新器。
AtomicLongFieldUpdater: 原子更新長整型字段的更新器。
AtomicStampedFieldUpdater: 原子更新帶有版本號的引用類型
AtomicReferenceFieldUpdater: 上面已經說過此處不在贅述。

如下所示，我們創建一個基礎類DataDemo，通過原子類CAS操作字段值進行自增操作。

public class TestAtomicIntegerFieldUpdater {


    private static Logger logger = LoggerFactory.getLogger(TestAtomicIntegerFieldUpdater.class);

    public static void main(String[] args) {
        TestAtomicIntegerFieldUpdater tIA = new TestAtomicIntegerFieldUpdater();
        tIA.doIt();
    }

    /**
     * 返回需要更新的整型字段更新器
     *
     * @param fieldName
     * @return
     */
    public AtomicIntegerFieldUpdater<DataDemo> updater(String fieldName) {
        return AtomicIntegerFieldUpdater.newUpdater(DataDemo.class, fieldName);
    }

    public void doIt() {
        DataDemo data = new DataDemo();
        // 修改公共變量，返回更新前的舊值 0
        AtomicIntegerFieldUpdater<DataDemo> updater = updater("publicVar");
        int oldVal = updater.getAndIncrement(data);
        logger.info("publicVar 更新前的值[{}] 更新后的值 [{}]", oldVal, data.publicVar);


        // 更新保護級別的變量
        AtomicIntegerFieldUpdater<DataDemo> protectedVarUpdater = updater("protectedVar");
        int oldProtectedVar = protectedVarUpdater.getAndAdd(data, 2);
        logger.info("protectedVar 更新前的值[{}] 更新后的值 [{}]", oldProtectedVar, data.protectedVar);


        // logger.info("privateVar = "+updater("privateVar").getAndAdd(data,2)); 私有變量會報錯

        /*
         * 下面報異常：must be integer
         * */
//        logger.info("integerVar = "+updater("integerVar").getAndIncrement(data));
        //logger.info("longVar = "+updater("longVar").getAndIncrement(data));
    }


    class DataDemo {

        // 公共且可見的publicVar
        public volatile int publicVar = 0;
        // 保護級別的protectedVar
        protected volatile int protectedVar = 4;
        // 私有變量
        private volatile int privateVar = 5;

        // final 不可變量
        public final int finalVar = 11;

        public volatile Integer integerVar = 19;
        public volatile Long longVar = 18L;

    }

}
public class TestAtomicIntegerFieldUpdater {


    private static Logger logger = LoggerFactory.getLogger(TestAtomicIntegerFieldUpdater.class);

    public static void main(String[] args) {
        TestAtomicIntegerFieldUpdater tIA = new TestAtomicIntegerFieldUpdater();
        tIA.doIt();
    }

    /**
     * 返回需要更新的整型字段更新器
     *
     * @param fieldName
     * @return
     */
    public AtomicIntegerFieldUpdater<DataDemo> updater(String fieldName) {
        return AtomicIntegerFieldUpdater.newUpdater(DataDemo.class, fieldName);
    }

    public void doIt() {
        DataDemo data = new DataDemo();
        // 修改公共變量，返回更新前的舊值 0
        AtomicIntegerFieldUpdater<DataDemo> updater = updater("publicVar");
        int oldVal = updater.getAndIncrement(data);
        logger.info("publicVar 更新前的值[{}] 更新后的值 [{}]", oldVal, data.publicVar);


        // 更新保護級別的變量
        AtomicIntegerFieldUpdater<DataDemo> protectedVarUpdater = updater("protectedVar");
        int oldProtectedVar = protectedVarUpdater.getAndAdd(data, 2);
        logger.info("protectedVar 更新前的值[{}] 更新后的值 [{}]", oldProtectedVar, data.protectedVar);


        // logger.info("privateVar = "+updater("privateVar").getAndAdd(data,2)); 私有變量會報錯

        /*
         * 下面報異常：must be integer
         * */
//        logger.info("integerVar = "+updater("integerVar").getAndIncrement(data));
        //logger.info("longVar = "+updater("longVar").getAndIncrement(data));
    }


    class DataDemo {

        // 公共且可見的publicVar
        public volatile int publicVar = 0;
        // 保護級別的protectedVar
        protected volatile int protectedVar = 4;
        // 私有變量
        private volatile int privateVar = 5;

        // final 不可變量
        public final int finalVar = 11;

        public volatile Integer integerVar = 19;
        public volatile Long longVar = 18L;

    }

}

通過上述代碼我們可以總結出CAS字段必須符合以下要求：

變量必須使用volatile保證可見性
必須是當前對象可以訪問到的類型才可進行操作‘
只能是實例變量而不是類變量，即不可以有static修飾符
包裝類也不行

七、CAS的ABA問題

CAS更新前會檢查值有沒有變化，如果沒有變化則認為沒人修改過，在進行更新操作。這種情況下，若我們A值修改為B，B再還原為A。這種修改再還原的操作，CAS是無法感知是否變化的，這就是所謂的ABA問題。

1.AtomicStampedReference源碼詳解

源碼如下所示，可以看到AtomicStampedReference解決ABA問題的方式是基于當前修改操作的時間戳和元引用值是否一致，若一直則進行數據更新

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;
    ....
    
    /**
      * expectedReference ：更新之前的原始引用值
      * newReference : 新值
      * expectedStamp : 預期時間戳
      * newStamp : 更新后的時間戳
      */
    public boolean compareAndSet(V   expectedReference,
                             V   newReference,
                             int expectedStamp,
                             int newStamp) {
        // 獲取當前的(元素值，版本號)對
        Pair<V> current = pair;
        return
            // 引用沒變
            expectedReference == current.reference &&
            // 版本號沒變
            expectedStamp == current.stamp &&
           //可以看到這個括號里面用了一個短路運算如果當前版本與新值一樣就說更新過，就不往下走CAS代碼了
            ((newReference == current.reference &&
           
            newStamp == current.stamp) ||
            // 構造新的Pair對象并CAS更新
            casPair(current, Pair.of(newReference, newStamp)));
    }

    private boolean casPair(Pair<V> cmp, Pair<V> val) {
        // 調用Unsafe的compareAndSwapObject()方法CAS更新pair的引用為新引用
        return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
    }

2.AtomicStampedReference解決ABA問題示例

代碼示例，我們下面就用other代碼模擬干擾現場，如果other現場先進行CAS更新再還原操作，那么main線程的版本號就會過時，CAS就會操作失敗

/**
 * ABA問題代碼示例
 */
public class AtomicStampedReferenceTest {
    private static AtomicStampedReference<Integer> atomicStampedRef =
            new AtomicStampedReference<>(1, 0);

    public static void main(String[] args) {
        Thread main = new Thread(() -> {
            System.out.println("操作線程" + Thread.currentThread() + ",初始值 a = " + atomicStampedRef.getReference());
            int stamp = atomicStampedRef.getStamp(); //獲取當前標識別
            try {
                Thread.sleep(1000); //等待1秒 ，以便讓干擾線程執行
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            boolean isCASSuccess = atomicStampedRef.compareAndSet(1, 2, stamp, stamp + 1);  //此時expectedReference未發生改變，但是stamp已經被修改了,所以CAS失敗
            System.out.println("操作線程" + Thread.currentThread() + ",CAS操作結果: " + isCASSuccess);
        }, "主操作線程");

        Thread other = new Thread(() -> {
            Thread.yield(); // 確保thread-main 優先執行
            atomicStampedRef.compareAndSet(1, 2, atomicStampedRef.getStamp(), atomicStampedRef.getStamp() + 1);
            System.out.println("操作線程" + Thread.currentThread() + ",【increment】 ,值 = " + atomicStampedRef.getReference());
            atomicStampedRef.compareAndSet(2, 1, atomicStampedRef.getStamp(), atomicStampedRef.getStamp() + 1);
            System.out.println("操作線程" + Thread.currentThread() + ",【decrement】 ,值 = " + atomicStampedRef.getReference());
        }, "干擾線程");

        main.start();
        other.start();
    }

}

3.AtomicMarkableReference解決對象ABA問題

AtomicMarkableReference，它不是維護一個版本號，而是維護一個boolean類型的標記，標記對象是否有修改，從而解決ABA問題。

public boolean weakCompareAndSet(V       expectedReference,
                                     V       newReference,
                                     boolean expectedMark,
                                     boolean newMark) {
        return compareAndSet(expectedReference, newReference,
                             expectedMark, newMark);
    }

八、常見面試題

1.CAS為什么比synchronized快(重點)

CAS工作原理是基于樂觀鎖且操作是原子性的，與synchronized的悲觀鎖(底層需要調用操作系統的mutex鎖)相比，效率也會相對高一些。

2.CAS是不是操作系統執行的？(重點)

不是，CAS是主要是通過處理器的指令來保證原子性的，在上面的講解中我們都知道CAS操作底層都是調用Unsafe的native修飾的方法，以AtomicInteger為例對應的底層的實現是Unsafe的compareAndSwapInt：

 public final native boolean compareAndSwapInt(Object o, long offset,
                                                  int expected,
                                                  int x);

對應的我們給出這段代碼的c語言實現，即位于:https://github.com/openjdk/jdk/blob/jdk8-b01/hotspot/src/share/vm/prims/unsafe.cpp的unsafe.cpp:

可以看到出去前兩個形參后續的參數與compareAndSwapInt列表一一對應，這段代碼執行CAS操作時，本質上就是調用cmpxchg指令(Compare and Exchange)，cmpxchg指令會判斷當前服務器是否是多核，如果是則講LOCK前綴保證cmpxchg操作的原子性，反之就不加Lock前綴直接執行比對后修改變量值這種樂觀鎖操作。

對應源碼如下，它首先獲取字段的偏移地址，然后傳入預期值e與原值比較，如果一致，則將新結果x寫入原子操作變量內存中：

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);
  //比較如果期望值e和當前字段存儲的值一樣，則講值更新為x
  return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
UNSAFE_END

3.CAS存在那些問題?

但即便如此CAS仍然存在兩個問題：

(1) 可能存在長時間CAS：如下代碼所示，這就是AtomicInteger底層的UNSAFE類如何進行CAS的具體代碼，可以看出這個CAS操作需要拿到volatile變量后在進行循環CAS才有可能成功這就很可能存在自旋循環，從而給CPU帶來很大的執行開銷。

public final int getAndAddInt(Object paramObject, long paramLong, int paramInt)
  {
    int i;
    do
     //獲取最新結果
      i = getIntVolatile(paramObject, paramLong);
      //通過cas自旋操作完成自增
    while (!compareAndSwapInt(paramObject, paramLong, i, i + paramInt));
    return i;
  }

(2) CAS只能對一個變量進行原子操作：為了解決這個問題，JDK 1.5之后通過AtomicReference使得變量可以封裝成一個對象進行操作

ABA問題：總所周知CAS就是比對當前值與舊值是否相等，在進行修改操作，假設我現在有一個變量值為A，我改為B，再還原為A，這樣操作變量值是沒變的？那么CAS也會成功不就不合理嗎？這就好比一個銀行儲戶想查詢概念轉賬記錄，如果轉賬一次記為1，如果按照ABA問題的邏輯，那么這個銀行賬戶轉賬記錄次數有可能會缺少。為了解決這個問題JDK 1.5提供了AtomicStampedReference，通過比對版本號在進行CAS操作，那么上述操作就會變為1A->2B->3A,由于版本追加，那么我們就能捕捉到當前變量的變化了。

4.AtomicInteger自增到10000后如何歸零

AtomicInteger atomicInteger=new AtomicInteger(10000);
atomicInteger.compareAndSet(10000, 0);

5.CAS 平時怎么用的，會有什么問題，為什么快，如果我用 for 循環代替 CAS 執行效率是一樣的嗎？(重點)

問題1：一些需要并發計數并實時監控的場景可以用到。

問題2： CAS存在問題：CAS是基于樂觀鎖機制，所以數據同步失敗就會原地自旋，在高并發場景下開銷很大，所以線程數很大的情況下不建議使用原子類。

問題3：用 for 循環代替 CAS 存在問題: 如果并發量大的話，自旋的線程多了就會導致性能瓶頸。

for 循環代替 CAS執行效率是否一樣:大概率是CAS快，原因如下:

CAS是native方法更接近底層
for循環為了保證線程安全可能會用到sync鎖或者Lock無論那種都需要上鎖和釋放的邏輯，相比CAS樂觀鎖來說開銷很大。

責任編輯：趙寧寧來源：寫代碼的SharkChili