CAS機制與原子性

學習材料來源於網絡
如有侵權，聯系刪除

概念

Compare and swap比較和交換。屬於硬件同步原語，處理器提供了基本內存操作的原子性保證。CAS操作需要輸入兩個數值;一個舊值A(期望操作前的值)和一個新值B，在操作期間先比較下舊值有沒有發生變化，如果沒有發生變化，才交換成新值，發生了變化則不交換。

JAVA中的sun.misc.Unsafe類，提供了compareAndSwapInt()和compareAndSwapLong()等幾個方法實現CAS.

示例1

import java.util.concurrent.atomic.AtomicInteger;

// 兩個線程，對 i 變量進行遞增操作
public class LockDemo {
//    volatile int i = 0;
    AtomicInteger i = new AtomicInteger(0);


    public void add() {
        // TODO xx00
//         i++;// 三個步驟
        i.incrementAndGet();
    }

    public static void main(String[] args) throws InterruptedException {
        LockDemo ld = new LockDemo();

        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    ld.add();
                }
            }).start();
        }
        Thread.sleep(2000L);
        System.out.println(ld.i);
    }
}

使用CAS操作，保證原子性

package icu.shaoyayu.multithreading.chapter2;

import sun.misc.Unsafe;

import java.lang.reflect.Field;

/**
 * @author shaoyayu
 * @date 2020/11/29
 * @E_Mail
 * @Version 1.0.0
 * @readme ：
 */
public class LockDemo1 {
    volatile int value = 0;
    // 直接操作內存，修改對象，數組內存....強大的API
    static Unsafe unsafe;
    private static long valueOffset;

    static {
        try {
            // 反射技術獲取unsafe值
            Field field = Unsafe.class.getDeclaredField("theUnsafe");
            field.setAccessible(true);
            unsafe = (Unsafe) field.get(null);

            // 獲取到 value 屬性偏移量（用於定於value屬性在內存中的具體地址）
            valueOffset = unsafe.objectFieldOffset(LockDemo1.class
                    .getDeclaredField("value"));

        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public void add() {
        // CAS + 循環 重試
        int current;
        do {
            // 操作耗時的話， 那么 線程就會占用大量的CPU執行時間
            current = unsafe.getIntVolatile(this, valueOffset);
        } while (!unsafe.compareAndSwapInt(this, valueOffset, current, current + 1));
        // 可能會失敗
    }

    public static void main(String[] args) throws InterruptedException {
        LockDemo1 ld = new LockDemo1();

        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    ld.add();
                }
            }).start();
        }
        Thread.sleep(2000L);
        System.out.println(ld.value);
    }
}

使用原子性操作的話可以保持對某個內存地址的原子性

但是這樣一直處於循環狀態中，會占用CPU資源，所有可以使用同步關鍵字，避免這種問題

J.U.C包內的原子操作封裝類

AtomicBoolean:原子更新布爾類型

AtomicInteger:原子更新整型

AtomicLong:原子更新長整型

示例2

package icu.shaoyayu.multithreading.chapter2;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author shaoyayu
 * @E_Mail
 * @Version 1.0.0
 * @readme ：
 */
public class LockDemo3 {

    //原子性操作Int提供的對象

    AtomicInteger value = new AtomicInteger(0);

    public void add() {
        // TODO xx00
        //
        value.incrementAndGet();
    }

    public static void main(String[] args) throws InterruptedException {
        LockDemo3 ld = new LockDemo3();

        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    ld.add();
                }
            }).start();
        }
        Thread.sleep(2000L);
        System.out.println(ld.value);
    }
}

源碼分析：

/ ** 
    *以原子方式將當前值增加1。 
    * 
    * @返回更新的值
    * /
public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}

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;
    }

處理上面舉例的數據類型外，還有

AtomicBoolean:原子更新布爾類型
AtomicInteger:原子更新整型
AtomicLong:原子更新長整型
AtomiclntegerArray:原子更新整型數組里的元素。
AtomicLongArray:原子更新長整型數組里的元素。
AtomicReferenceArray:原子更新引用類型數組里的元素。
AtomiclntegerFieldUpdater:原子更新整型的字段的更新器。
AtomicLongFieldUpdater:原子更新長整型字段的更新器。
AtomicReferenceFieldUpdater:原子更新引用類型里的字段。
AtomicReference:原子更新引用類型。
AtomicStampedReference:原子更新帶有版本號的引用類型
AtomicMarkableReference:原子更新帶有標記位的引用類型。
1.8更新
更新器:DoubleAccumulator、LongAccumulator
計數器:DoubleAdder、LongAdder
計數器增強版，高並發下性能更好
頻繁更新但不太頻繁讀取的匯總統計信息時使用分成多個操作單元，不同線程更新不同的單元只有需要匯總的時候才計算所有單元的操作

計數器

為多個線程共享的變量創建多個內存地址供多個線程操作，最后一致性匯總給線程。

package icu.shaoyayu.multithreading.chapter2;

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
/**
 * @author shaoyayu
 * @date 2020/11/29
 * @E_Mail
 * @Version 1.0.0
 * @readme ：
 */
// 測試用例： 同時運行2秒，檢查誰的次數最多
public class LongAdderDemo {
    private long count = 0;

    // 同步代碼塊的方式
    public void testSync() throws InterruptedException {
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                long starttime = System.currentTimeMillis();
                while (System.currentTimeMillis() - starttime < 2000) { // 運行兩秒
                    synchronized (this) {
                        ++count;
                    }
                }
                long endtime = System.currentTimeMillis();
                System.out.println("SyncThread spend:" + (endtime - starttime) + "ms" + " v" + count);
            }).start();
        }
    }

    // Atomic方式
    private AtomicLong acount = new AtomicLong(0L);

    public void testAtomic() throws InterruptedException {
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                long starttime = System.currentTimeMillis();
                while (System.currentTimeMillis() - starttime < 2000) { // 運行兩秒
                    acount.incrementAndGet(); // acount++;
                }
                long endtime = System.currentTimeMillis();
                System.out.println("AtomicThread spend:" + (endtime - starttime) + "ms" + " v-" + acount.incrementAndGet());
            }).start();
        }
    }

    // LongAdder 方式
    private LongAdder lacount = new LongAdder();
    public void testLongAdder() throws InterruptedException {
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                long starttime = System.currentTimeMillis();
                while (System.currentTimeMillis() - starttime < 2000) { // 運行兩秒
                    lacount.increment();
                }
                long endtime = System.currentTimeMillis();
                System.out.println("LongAdderThread spend:" + (endtime - starttime) + "ms" + " v-" + lacount.sum());
            }).start();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        LongAdderDemo demo = new LongAdderDemo();
        demo.testSync();
        demo.testAtomic();
        demo.testLongAdder();
    }
}

運行結果：

SyncThread spend:2000ms v22305402
SyncThread spend:2005ms v22305404
SyncThread spend:2005ms v22305404
AtomicThread spend:2000ms v-71491578
AtomicThread spend:2000ms v-71492707
LongAdderThread spend:2000ms v-116795972
LongAdderThread spend:2000ms v-116796436
AtomicThread spend:2000ms v-71951742
LongAdderThread spend:2000ms v-118008296

源碼：

/**
* Adds the given value.
*
* @param x the value to add
*/
public void add(long x) {
    Cell[] as; long b, v; int m; Cell a;
    if ((as = cells) != null || !casBase(b = base, b + x)) {
        boolean uncontended = true;
        if (as == null || (m = as.length - 1) < 0 ||
            (a = as[getProbe() & m]) == null ||
            !(uncontended = a.cas(v = a.value, v + x)))
            longAccumulate(x, null, uncontended);
    }
}

/ ** 
    *返回當前總和。返回的值為<em> NOT </ em> 
    *原子快照；在沒有並發
    *更新的情況下調用將返回准確的結果，但是
    *在計算總和時發生的並發更新
    *可能不會被合並。 
    * 
    * @返回總和
    * /
public long sum() {
    Cell[] as = cells; Cell a;
    long sum = base;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

適合頻繁更新，但不適合頻繁讀取

支持CAS的其他自定義方法的增強版

package icu.shaoyayu.multithreading.chapter2;

import java.util.concurrent.atomic.LongAccumulator;
import java.util.function.LongBinaryOperator;
/**
 * @author shaoyayu
 * @E_Mail
 * @Version 1.0.0
 * @readme ：
 */
// LongAdder增強版，處理累加之外，可以自行定義其他計算
public class LongAccumulatorDemo {
    public static void main(String[] args) throws InterruptedException {
        LongAccumulator accumulator = new LongAccumulator(new LongBinaryOperator() {
            @Override
            public long applyAsLong(long left, long right) {
                // 返回最大值，這就是自定義的計算
                return left > right ? left : right;
            }
        }, 0);
        // 1000個線程
        for (int i = 0; i < 1000; i++) {
            int finalI = i;
            new Thread(() -> {
                accumulator.accumulate(finalI); // 此處實際就是執行上面定義的操作
            }).start();
        }
        Thread.sleep(2000L);
        System.out.println(accumulator.longValue()); // 打印出結果
    }
}

CAS的三個問題

1.循環+CAS，自旋的實現讓所有線程都處於高頻運行，爭搶CPU執行時間的狀態。如果操作長時間不成功，會帶來很大的CPU資源消耗。

2．僅針對單個變量的操作，不能用於多個變量來實現原子操作。

3.ABA問題。(此處結合代碼理解)）。