摘要：我們繼續看代碼的意思是這個是一段內嵌匯編代碼。也就是在語言中使用匯編代碼。就是匯編版的比較并交換。就是保證在多線程情況下，不阻塞線程的填充和消費。微觀上看匯編的是實現操作系統級別的原子操作的基石。

原文地址：https://www.xilidou.com/2018/02/01/java-cas/

CAS 是現代操作系統，解決并發問題的一個重要手段，最近在看 eureka 的源碼的時候。遇到了很多 CAS 的操作。今天就系統的回顧一下 Java 中的CAS。

閱讀這篇文章你將會了解到：

什么是 CAS

CAS 實現原理是什么？

CAS 在現實中的應用

自旋鎖

原子類型

限流器

CAS 的缺點

CAS: 全稱Compare and swap，字面意思:”比較并交換“，一個 CAS 涉及到以下操作：

我們假設內存中的原數據V，舊的預期值A，需要修改的新值B。

比較 A 與 V 是否相等。（比較）
如果比較相等，將 B 寫入 V。（交換）
返回操作是否成功。

當多個線程同時對某個資源進行CAS操作，只能有一個線程操作成功，但是并不會阻塞其他線程,其他線程只會收到操作失敗的信號。可見 CAS 其實是一個樂觀鎖。

跟隨AtomInteger的代碼我們一路往下，就能發現最終調用的是 sum.misc.Unsafe 這個類。看名稱 Unsafe 就是一個不安全的類，這個類是利用了 Java 的類和包在可見性的的規則中的一個恰到好處處的漏洞。Unsafe 這個類為了速度，在Java的安全標準上做出了一定的妥協。

再往下尋找我們發現 Unsafe的compareAndSwapInt 是 Native 的方法：

public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);

也就是說，這幾個 CAS 的方法應該是使用了本地的方法。所以這幾個方法的具體實現需要我們自己去 jdk 的源碼中搜索。

于是我下載一個 OpenJdk 的源碼繼續向下探索，我們發現在 /jdk9u/hotspot/src/share/vm/unsafe.cpp 中有這樣的代碼：

{CC "compareAndSetInt",   CC "(" OBJ "J""I""I"")Z",  FN_PTR(Unsafe_CompareAndSetInt)},

這個涉及到，JNI 的調用，感興趣的同學可以自行學習。我們搜索 Unsafe_CompareAndSetInt后發現:

UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSetInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x)) {
  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

最終我們終于看到了核心代碼 Atomic::cmpxchg。

繼續向底層探索，在文件java/jdk9u/hotspot/src/os_cpu/linux_x86/vm/atomic_linux_x86.hpp有這樣的代碼:

inline jint     Atomic::cmpxchg    (jint     exchange_value, volatile jint*     dest, jint     compare_value, cmpxchg_memory_order order) {
  int mp = os::is_MP();
  __asm__ volatile (LOCK_IF_MP(%4) "cmpxchgl %1,(%3)"
                    : "=a" (exchange_value)
                    : "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
                    : "cc", "memory");
  return exchange_value;
}

我們通過文件名可以知道，針對不同的操作系統,JVM 對于 Atomic::cmpxchg 應該有不同的實現。由于我們服務基本都是使用的是64位linux，所以我們就看看linux_x86 的實現。

我們繼續看代碼：

__asm__ 的意思是這個是一段內嵌匯編代碼。也就是在 C 語言中使用匯編代碼。

這里的 volatile和 JAVA 有一點類似，但不是為了內存的可見性，而是告訴編譯器對訪問該變量的代碼就不再進行優化。

LOCK_IF_MP(%4) 的意思就比較簡單，就是如果操作系統是多核的，那就增加一個 LOCK。

cmpxchgl 就是匯編版的“比較并交換”。但是我們知道比較并交換，有三個步驟，不是原子的。所以在多核情況下加一個 LOCK，由CPU硬件保證他的原子性。

我們再看看 LOCK 是怎么實現的呢？我們去Intel的官網上看看，可以知道LOCK在的早期實現是直接將 cup 的總線阻塞，這樣的實現可見效率是很低下的。后來優化為X86 cpu 有鎖定一個特定內存地址的能力，當這個特定內存地址被鎖定后，它就可以阻止其他的系統總線讀取或修改這個內存地址。

關于 CAS 的底層探索我們就到此為止。我們總結一下 JAVA 的 cas 是怎么實現的：

java 的 cas 利用的的是 unsafe 這個類提供的 cas 操作。

unsafe 的cas 依賴了的是 jvm 針對不同的操作系統實現的 Atomic::cmpxchg

Atomic::cmpxchg 的實現使用了匯編的 cas 操作，并使用 cpu 硬件提供的 lock信號保證其原子性

了解了 CAS 的原理我們繼續就看看 CAS 的應用：

public class SpinLock {

  private AtomicReference sign =new AtomicReference<>();

  public void lock(){
    Thread current = Thread.currentThread();
    while(!sign .compareAndSet(null, current)){
    }
  }

  public void unlock (){
    Thread current = Thread.currentThread();
    sign .compareAndSet(current, null);
  }
}

所謂自旋鎖，我覺得這個名字相當的形象，在lock()的時候，一直while()循環，直到 cas 操作成功為止。

    public final int getAndAddInt(Object var1, long var2, int var4) {
        int var5;
        do {
            var5 = this.getIntVolatile(var1, var2);
        } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

        return var5;
    }

與自旋鎖有異曲同工之妙，就是一直while，直到操作成功為止。

所謂令牌桶限流器，就是系統以恒定的速度向桶內增加令牌。每次請求前從令牌桶里面獲取令牌。如果獲取到令牌就才可以進行訪問。當令牌桶內沒有令牌的時候，拒絕提供服務。我們來看看 eureka 的限流器是如何使用 CAS 來維護多線程環境下對 token 的增加和分發的。

public class RateLimiter {

    private final long rateToMsConversion;

    private final AtomicInteger consumedTokens = new AtomicInteger();
    private final AtomicLong lastRefillTime = new AtomicLong(0);

    @Deprecated
    public RateLimiter() {
        this(TimeUnit.SECONDS);
    }

    public RateLimiter(TimeUnit averageRateUnit) {
        switch (averageRateUnit) {
            case SECONDS:
                rateToMsConversion = 1000;
                break;
            case MINUTES:
                rateToMsConversion = 60 * 1000;
                break;
            default:
                throw new IllegalArgumentException("TimeUnit of " + averageRateUnit + " is not supported");
        }
    }

    //提供給外界獲取 token 的方法
    public boolean acquire(int burstSize, long averageRate) {
        return acquire(burstSize, averageRate, System.currentTimeMillis());
    }

    public boolean acquire(int burstSize, long averageRate, long currentTimeMillis) {
        if (burstSize <= 0 || averageRate <= 0) { // Instead of throwing exception, we just let all the traffic go
            return true;
        }

        //添加token
        refillToken(burstSize, averageRate, currentTimeMillis);

        //消費token
        return consumeToken(burstSize);
    }

    private void refillToken(int burstSize, long averageRate, long currentTimeMillis) {
        long refillTime = lastRefillTime.get();
        long timeDelta = currentTimeMillis - refillTime;

        //根據頻率計算需要增加多少 token
        long newTokens = timeDelta * averageRate / rateToMsConversion;
        if (newTokens > 0) {
            long newRefillTime = refillTime == 0
                    ? currentTimeMillis
                    : refillTime + newTokens * rateToMsConversion / averageRate;

            // CAS 保證有且僅有一個線程進入填充
            if (lastRefillTime.compareAndSet(refillTime, newRefillTime)) {
                while (true) {
                    int currentLevel = consumedTokens.get();
                    int adjustedLevel = Math.min(currentLevel, burstSize); // In case burstSize decreased
                    int newLevel = (int) Math.max(0, adjustedLevel - newTokens);
                    // while true 直到更新成功為止
                    if (consumedTokens.compareAndSet(currentLevel, newLevel)) {
                        return;
                    }
                }
            }
        }
    }

    private boolean consumeToken(int burstSize) {
        while (true) {
            int currentLevel = consumedTokens.get();
            if (currentLevel >= burstSize) {
                return false;
            }

            // while true 直到沒有token 或者 獲取到為止
            if (consumedTokens.compareAndSet(currentLevel, currentLevel + 1)) {
                return true;
            }
        }
    }

    public void reset() {
        consumedTokens.set(0);
        lastRefillTime.set(0);
    }
}

所以梳理一下 CAS 在令牌桶限流器的作用。就是保證在多線程情況下，不阻塞線程的填充token 和消費token。

通過上面的三個應用我們歸納一下 CAS 的應用場景：

CAS 的使用能夠避免線程的阻塞。

多數情況下我們使用的是 while true 直到成功為止。

ABA 的問題，就是一個值從A變成了B又變成了A，使用CAS操作不能發現這個值發生變化了，處理方式是可以使用攜帶類似時間戳的版本AtomicStampedReference

性能問題，我們使用時大部分時間使用的是 while true 方式對數據的修改，直到成功為止。優勢就是相應極快，但當線程數不停增加時，性能下降明顯，因為每個線程都需要執行，占用CPU時間。

CAS 是整個編程重要的思想之一。整個計算機的實現中都有CAS的身影。微觀上看匯編的 CAS 是實現操作系統級別的原子操作的基石。從編程語言角度來看 CAS 是實現多線程非阻塞操作的基石。宏觀上看，在分布式系統中，我們可以使用 CAS 的思想利用類似Redis的外部存儲，也能實現一個分布式鎖。

從某個角度來說架構就將微觀的實現放大，或者底層思想就是將宏觀的架構進行微縮。計算機的思想是想通的，所以說了解底層的實現可以提升架構能力，提升架構的能力同樣可加深對底層實現的理解。計算機知識浩如煙海，但是套路有限。抓住基礎的幾個套路突破，從思想和思維的角度學習計算機知識。不要將自己的精力花費在不停的追求新技術的腳步上，跟隨‘start guide line’只能寫一個demo，所得也就是一個demo而已。

停下腳步，回顧基礎和經典或許對于技術的提升更大一些。

希望這篇文章對大家有所幫助。

徒手擼框架系列文章地址：

徒手擼框架--高并發環境下的請求合并
徒手擼框架--實現IoC
徒手擼框架--實現Aop

歡迎關注我的微信公眾號