Java並行程式設計之原子性-Atomic的使用

2022-03-15 19:00:35

執行緒安全

當多個執行緒存取某個類時，不管執行時環境採用何種排程方式或者這些程序將如何交替執行，並且在主調程式碼中不需要任何額外的同步或協調，這個類都能表現出正確的行為，那麼就稱這個類時執行緒安全的。

執行緒安全主要體現在以下三個方面

原子性：提供了互斥存取，同一時刻只能有一個執行緒對它進行操作
可見性：一個執行緒對主記憶體的修改可以及時的被其他執行緒觀察到
有序性：一個執行緒觀察其他執行緒中的指令執行順序，由於指令重排序的存在，該觀察結果一般雜亂無序

JUC中的Atomic包詳解

Atomic包中提供了很多Atomicxxx的類：

它們都是CAS（compareAndSwap）來實現原子性。

先寫一個簡單範例如下：

@Slf4j
public class AtomicExample1 { 
    // 請求總數
    public static int clientTotal = 5000; 
    // 同時並行執行的執行緒數
    public static int threadTotal = 200; 
    public static AtomicInteger count = new AtomicInteger(0); 
    public static void main(String[] args) throws Exception {

        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal ; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    add();
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("count:{}", count.get());
    }
 
    private static void add() {
        count.incrementAndGet();
    }
}

可以發下每次的執行結果總是我們想要的預期結果5000。說明該計數方法是執行緒安全的。

我們檢視下count.incrementAndGet()方法，它的第一個引數為物件本身，第二個引數為valueOffset是用來記錄value本身在記憶體的編譯地址的，這個記錄，也主要是為了在更新操作在記憶體中找到value的位置，方便比較，第三個引數為常數1

public class AtomicInteger extends Number implements java.io.Serializable {
    private static final long serialVersionUID = 6214790243416807050L;
 
    // setup to use Unsafe.compareAndSwapInt for updates
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final long valueOffset;
 
    static {
        try {
            valueOffset = unsafe.objectFieldOffset
                (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }
 
    private volatile int value; 
 
    ... 此處省略多個方法...
 
    /**
     * Atomically increments by one the current value.
     *
     * @return the updated value
     */
    public final int incrementAndGet() {
        return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
    }
}

AtomicInteger原始碼裡使用了一個Unsafe的類,它提供了一個getAndAddInt的方法，我們繼續點看檢視它的原始碼：

public final class Unsafe {
    private static final Unsafe theUnsafe;
 
    ....此處省略很多方法及成員變數.... 
 
 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;
    } 
 
 public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5); 
 public native int getIntVolatile(Object var1, long var2);
}

可以看到這裡使用了一個do while語句來做主體實現的。而在while語句裡它的核心是呼叫了一個compareAndSwapInt()的方法。它是一個native方法，它是一個底層的方法，不是使用Java來實現的。

假設我們要執行0+1=0的操作，下面是單執行緒情況下各引數的值：

更新後：

compareAndSwapInt()方法的第一個引數（var1）是當前的物件，就是程式碼範例中的count。此時它的值為0（期望值）。第二個值（var2）是傳遞的valueOffset值，它的值為12。第三個引數（var4）就為常數1。方法中的變數引數（var5）是根據引數一和引數二valueOffset，呼叫底層getIntVolatile方法得到的值，此時它的值為0 。compareAndSwapInt()想要達到的目標是對於count這個物件，如果當前的期望值var1裡的value跟底層的返回的值（var5）相同的話，那麼把它更新成var5+var4這個值。不同的話重新迴圈取期望值（var5）直至當前值與期望值相同才做更新。compareAndSwap方法的核心也就是我們通常所說的CAS。

Atomic包下其他的類如AtomicLong等的實現原理基本與上述一樣。

這裡再介紹下LongAdder這個類，通過上述的分析，我們已經知道了AtomicLong使用CAS：在一個死迴圈內不斷嘗試修改目標值直到修改成功。如果在競爭不激烈的情況下，它修改成功概率很高。反之，如果在競爭激烈的情況下，修改失敗的概率會很高，它就會進行多次的迴圈嘗試，因此效能會受到一些影響。

對於普通型別的long和double變數，jvm允許將64位元的讀操作或寫操作拆成兩個32位元的操作。LongAdder的核心思想是將熱點資料分離，它可以將AtomicLong內部核心資料value分離成一個陣列，每個執行緒存取時通過hash等演演算法對映到其中一個數位進行計數。而最終的計數結果則為這個陣列的求和累加，其中熱點資料value，它會被分離成多個單元的cell，每個cell獨自維護內部的值,當前物件的實際值由所有的cell累計合成。這樣,熱點就進行了有效的分離,提高了並行度。LongAdder相當於在AtomicLong的基礎上將單點的更新壓力分散到各個節點上，在低並行的時候對base的直接更新可以很好的保障跟Atomic的效能基本一致。而在高並行的時候，通過分散提高了效能。但是如果在統計的時候有並行更新，可能會導致統計的資料有誤差。

在實際高並行計數的時候，可以優先使用LongAdder。在低並行度或者需要準確數值的時候可以優先使用AtomicLong，這樣反而效率更高。

下面簡單的演示下Atomic包下AtomicReference簡單的用法：

@Slf4j
public class AtomicExample4 { 
    private static AtomicReference<Integer> count = new AtomicReference<>(0); 
    public static void main(String[] args) {
        count.compareAndSet(0, 2); 
        count.compareAndSet(0, 1); 
        log.info("count:{}", count.get());
    }
}

compareAndSet()分別傳入的是預期值跟更新值，只有當預期值跟當前值相等時，才會將值更新為更新值；

上面的第一個方法可以將值更新為2，而第二個步中無法將值更新為1。

下面簡單介紹下AtomicIntegerFieldUpdater 用法（利用原子性去更新某個類的範例）：

@Slf4j
public class AtomicExample5 { 
    private static AtomicIntegerFieldUpdater<AtomicExample5> updater =
            AtomicIntegerFieldUpdater.newUpdater(AtomicExample5.class, "count");
 
    @Getter
    private volatile int count = 100; 
    public static void main(String[] args) { 
        AtomicExample5 example5 = new AtomicExample5();
 
        if (updater.compareAndSet(example5, 100, 120)) {
            log.info("update success 1, {}", example5.getCount());
        }
 
        if (updater.compareAndSet(example5, 100, 120)) {
            log.info("update success 2, {}", example5.getCount());
        } else {
            log.info("update failed, {}", example5.getCount());
        }
    }
}

它可以更新某個類中指定成員變數的值。

注意：修改的成員變數需要用volatile關鍵字來修飾，並且不能是static描述的欄位。

AtomicStampReference這個類它的核心是要解決CAS的ABA問題（CAS操作的時候，其他執行緒將變數的值A改成了B，接著又改回了A，等執行緒使用期望值A與當前變數進行比較的時候，發現A變數沒有變，於是CAS就將A值進行了交換操作。

實際上該值已經被其他執行緒改變過）。

ABA問題的解決思路就是每次變數變更的時候，就將版本號加一。

看一下它的一個核心方法compareAndSet()：

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);
        }
    }
 
   ... 此處省略多個方法 ....
 
   public boolean compareAndSet(V   expectedReference,
                                 V   newReference,
                                 int expectedStamp,
                                 int newStamp) {
        Pair<V> current = pair;
        return
            expectedReference == current.reference &&
            expectedStamp == current.stamp &&
            ((newReference == current.reference &&
              newStamp == current.stamp) ||
             casPair(current, Pair.of(newReference, newStamp)));
    }
}

可以看到它多了一個stamp的比較，stamp的值是由每次更新的時候進行維護的。

再介紹下AtomicLongArray，它維護了一個陣列。在該陣列下，我們可以選擇性的已原子性操作更新某個索引對應的值。

public class AtomicLongArray implements java.io.Serializable {
    private static final long serialVersionUID = -2308431214976778248L;
 
    private static final Unsafe unsafe = Unsafe.getUnsafe();
 
    ...此處省略....
 
 
    /**
     * Atomically sets the element at position {@code i} to the given value
     * and returns the old value.
     *
     * @param i the index
     * @param newValue the new value
     * @return the previous value
     */
    public final long getAndSet(int i, long newValue) {
        return unsafe.getAndSetLong(array, checkedByteOffset(i), newValue);
    }
 
    /**
     * Atomically sets the element at position {@code i} to the given
     * updated value if the current value {@code ==} the expected value.
     *
     * @param i the index
     * @param expect the expected value
     * @param update the new value
     * @return {@code true} if successful. False return indicates that
     * the actual value was not equal to the expected value.
     */
    public final boolean compareAndSet(int i, long expect, long update) {
        return compareAndSetRaw(checkedByteOffset(i), expect, update);
    }
}

最後再寫一個AtomcBoolean的簡單使用：

@Slf4j
public class AtomicExample6 { 
    private static AtomicBoolean isHappened = new AtomicBoolean(false);
 
    // 請求總數
    public static int clientTotal = 5000;
 
    // 同時並行執行的執行緒數
    public static int threadTotal = 200;
 
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal ; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    test();
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("isHappened:{}", isHappened.get());
    }
 
    private static void test() {
        if (isHappened.compareAndSet(false, true)) {
            log.info("execute");
        }
    }
}

總結

以上就是Atomic包的基本原理及主要的使用方法。它是使用CAS來保證原子性操作，從而達到執行緒安全的目的。

僅為個人經驗，希望能給大家一個參考，也希望大家多多支援it145.com。

Java並行程式設計之原子性-Atomic的使用

目錄

執行緒安全

執行緒安全主要體現在以下三個方面

JUC中的Atomic包詳解

總結

熱門文章