基礎篇：JAVA原子組件和同步組件

2021 年 1 月 6 日
筆記
AQS, cas, JAVA, JAVA基礎篇, 同步, 同步鎖, 面試

前言

在使用多執行緒並發編程的時，經常會遇到對共享變數修改操作。此時我們可以選擇ConcurrentHashMap，ConcurrentLinkedQueue來進行安全地存儲數據。但如果單單是涉及狀態的修改，執行緒執行順序問題，使用Atomic開頭的原子組件或者ReentrantLock、CyclicBarrier之類的同步組件，會是更好的選擇，下面將一一介紹它們的原理和用法

原子組件的實現原理CAS
AtomicBoolean、AtomicIntegerArray等原子組件的用法、
同步組件的實現原理
ReentrantLock、CyclicBarrier等同步組件的用法

關注公眾號，一起交流，微信搜一搜: 潛行前行

原子組件的實現原理CAS

cas的底層實現可以看下之前寫的一篇文章:詳解鎖原理，synchronized、volatile+cas底層實現

應用場景

可用來實現變數、狀態在多執行緒下的原子性操作
可用於實現同步鎖(ReentrantLock)

原子組件

原子組件的原子性操作是靠使用cas來自旋操作volatile變數實現的
volatile的類型變數保證變數被修改時，其他執行緒都能看到最新的值
cas則保證value的修改操作是原子性的，不會被中斷

基本類型原子類

AtomicBoolean　//布爾類型
AtomicInteger　//正整型數類型
AtomicLong	  //長整型類型

使用示例

public static void main(String[] args) throws Exception {
    AtomicBoolean atomicBoolean = new AtomicBoolean(false);
    //非同步執行緒修改atomicBoolean
    CompletableFuture<Void> future = CompletableFuture.runAsync(() ->{
        try {
            Thread.sleep(1000); //保證非同步執行緒是在主執行緒之後修改atomicBoolean為false
            atomicBoolean.set(false);
        }catch (Exception e){
            throw new RuntimeException(e);
        }
    });
    atomicBoolean.set(true);
    future.join();
    System.out.println("boolean value is:"+atomicBoolean.get());
}
---------------輸出結果------------------
boolean value is:false

引用類原子類

AtomicReference
//加時間戳版本的引用類原子類
AtomicStampedReference
//相當於AtomicStampedReference，AtomicMarkableReference關心的是
//變數是否還是原來變數，中間被修改過也無所謂
AtomicMarkableReference

AtomicReference的源碼如下，它內部定義了一個volatile V value，並藉助VarHandle(具體子類是FieldInstanceReadWrite)實現原子操作，MethodHandles會幫忙計算value在類的偏移位置，最後在VarHandle調用Unsafe.public final native boolean compareAndSetReference(Object o, long offset, Object expected, Object x)方法原子修改對象的屬性

public class AtomicReference<V> implements java.io.Serializable {
    private static final long serialVersionUID = -1848883965231344442L;
    private static final VarHandle VALUE;
    static {
        try {
            MethodHandles.Lookup l = MethodHandles.lookup();
            VALUE = l.findVarHandle(AtomicReference.class, "value", Object.class);
        } catch (ReflectiveOperationException e) {
            throw new ExceptionInInitializerError(e);
        }
    }
    private volatile V value;
    ....

ABA問題

執行緒X準備將變數的值從A改為B，然而這期間執行緒Y將變數的值從A改為C，然後再改為A；最後執行緒X檢測變數值是A，並置換為B。但實際上，A已經不再是原來的A了
解決方法，是把變數定為唯一類型。值可以加上版本號，或者時間戳。如加上版本號，執行緒Y的修改變為A1->B2->A3，此時執行緒X再更新則可以判斷出A1不等於A3
AtomicStampedReference的實現和AtomicReference差不多，不過它原子修改的變數是volatile Pair<V> pair;，Pair是其內部類。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;

如果我們不關心變數在中間過程是否被修改過，而只是關心當前變數是否還是原先的變數，則可以使用AtomicMarkableReference
AtomicStampedReference的使用示例

public class Main {
    public static void main(String[] args) throws Exception {
        Test old = new Test("hello"), newTest = new Test("world");
        AtomicStampedReference<Test> reference = new AtomicStampedReference<>(old, 1);
        reference.compareAndSet(old, newTest,1,2);
        System.out.println("對象："+reference.getReference().name+";版本號："+reference.getStamp());
    }
}
class Test{
    Test(String name){ this.name = name; }
    public String name;
}
---------------輸出結果------------------
對象：world;版本號：2

數組原子類

AtomicIntegerArray	　//整型數組
AtomicLongArray		　//長整型數組
AtomicReferenceArray	//引用類型數組

數組原子類內部會初始一個final的數組，它把整個數組當做一個對象，然後根據下標index計演算法元素偏移量，再調用UNSAFE.compareAndSetReference進行原子操作。數組並沒被volatile修飾，為了保證元素類型在不同執行緒的可見，獲取元素使用到了UNSAFEpublic native Object getReferenceVolatile(Object o, long offset)方法來獲取實時的元素值
使用示例

//元素默認初始化為0
AtomicIntegerArray array = new AtomicIntegerArray(2);
// 下標為０的元素，期待值是0，更新值是１
array.compareAndSet(0,0,1);
System.out.println(array.get(0));
---------------輸出結果------------------
1

屬性原子更新類

AtomicIntegerFieldUpdater　
AtomicLongFieldUpdater
AtomicReferenceFieldUpdater

如果操作對象是某一類型的屬性，可以使用AtomicIntegerFieldUpdater原子更新，不過類的屬性需要定義成volatile修飾的變數，保證該屬性在各個執行緒的可見性，否則會報錯
使用示例

public class Main {
    public static void main(String[] args) {
        AtomicReferenceFieldUpdater<Test,String> fieldUpdater = AtomicReferenceFieldUpdater.newUpdater(Test.class,String.class,"name");
        Test test = new Test("hello world");
        fieldUpdater.compareAndSet(test,"hello world","siting");
        System.out.println(fieldUpdater.get(test));
        System.out.println(test.name);
    }
}
class Test{
    Test(String name){ this.name = name; }
    public volatile String name;
}
---------------輸出結果------------------
siting
siting

累加器

Striped64
LongAccumulator
LongAdder
//accumulatorFunction：運算規則，identity：初始值
public LongAccumulator(LongBinaryOperator accumulatorFunction,long identity)

LongAccumulator和LongAdder都繼承於Striped64，Striped64的主要思想是和ConcurrentHashMap有點類似，分段計算，單個變數計算並發性能慢時，我們可以把數學運算分散在多個變數，而需要計算總值時，再一一累加起來
LongAdder相當於LongAccumulator一個特例實現
LongAccumulator的示例

public static void main(String[] args) throws Exception {
    LongAccumulator accumulator = new LongAccumulator(Long::sum, 0);
    for(int i=0;i<100000;i++){
        CompletableFuture.runAsync(() -> accumulator.accumulate(1));
    }
    Thread.sleep(1000); //等待全部CompletableFuture執行緒執行完成，再獲取
    System.out.println(accumulator.get());
}
---------------輸出結果------------------
100000

同步組件的實現原理

java的多數同步組件會在內部維護一個狀態值，和原子組件一樣，修改狀態值時一般也是通過cas來實現。而狀態修改的維護工作被Doug Lea抽象出AbstractQueuedSynchronizer(AQS)來實現
AQS的原理可以看下之前寫的一篇文章:詳解鎖原理，synchronized、volatile+cas底層實現

同步組件

ReentrantLock、ReentrantReadWriteLock

ReentrantLock、ReentrantReadWriteLock都是基於AQS(AbstractQueuedSynchronizer)實現的。因為它們有公平鎖和非公平鎖的區分，因此沒直接繼承AQS，而是使用內部類去繼承，公平鎖和非公平鎖各自實現AQS，ReentrantLock、ReentrantReadWriteLock再藉助內部類來實現同步
ReentrantLock的使用示例

ReentrantLock lock = new ReentrantLock();
if(lock.tryLock()){
    //業務邏輯
    lock.unlock();
}

ReentrantReadWriteLock的使用示例

public static void main(String[] args) throws Exception {
    ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    if(lock.readLock().tryLock()){ //讀鎖
        //業務邏輯
        lock.readLock().unlock();
    }
    if(lock.writeLock().tryLock()){ //寫鎖
        //業務邏輯
        lock.writeLock().unlock();
    }
}

Semaphore實現原理和使用場景

Semaphore和ReentrantLock一樣，也有公平和非公平競爭鎖的策略，一樣也是通過內部類繼承AQS來實現同步
通俗解釋：假設有一口井，最多有三個人的位置打水。每有一個人打水，則需要佔用一個位置。當三個位置全部佔滿時，第四個人需要打水，則要等待前三個人中一個離開打水位，才能繼續獲取打水的位置
使用示例

public static void main(String[] args) throws Exception {
    Semaphore semaphore = new Semaphore(2);
    for (int i = 0; i < 3; i++)
        CompletableFuture.runAsync(() -> {
            try {
                System.out.println(Thread.currentThread().toString() + " start ");
                if(semaphore.tryAcquire(1)){
                    Thread.sleep(1000);
                    semaphore.release(1);
                    System.out.println(Thread.currentThread().toString() + " 無阻塞結束 ");
                }else {
                    System.out.println(Thread.currentThread().toString() + " 被阻塞結束 ");
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    //保證CompletableFuture 執行緒被執行，主執行緒再結束
    Thread.sleep(2000);
}
---------------輸出結果------------------
Thread[ForkJoinPool.commonPool-worker-19,5,main] start 
Thread[ForkJoinPool.commonPool-worker-5,5,main] start 
Thread[ForkJoinPool.commonPool-worker-23,5,main] start 
Thread[ForkJoinPool.commonPool-worker-23,5,main] 被阻塞結束 
Thread[ForkJoinPool.commonPool-worker-5,5,main] 無阻塞結束 
Thread[ForkJoinPool.commonPool-worker-19,5,main] 無阻塞結束

可以看出三個執行緒，因為訊號量設定為２，第三個執行緒是無法獲取資訊成功的，會列印阻塞結束

CountDownLatch實現原理和使用場景

CountDownLatch也是靠AQS實現的同步操作
通俗解釋：玩遊戲時，假如主線任務需要靠完成五個小任務，主線任務才能繼續進行時。此時可以用CountDownLatch，主線任務阻塞等待，每完成一小任務，就done一次計數，直到五個小任務全部被執行才能觸發主線
使用示例

public static void main(String[] args) throws Exception {
    CountDownLatch count = new CountDownLatch(2);
    for (int i = 0; i < 2; i++)
        CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);
                System.out.println(" CompletableFuture over ");
                count.countDown();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    //等待CompletableFuture執行緒的完成
    count.await();
    System.out.println(" main over ");
}
---------------輸出結果------------------
 CompletableFuture over 
 CompletableFuture over 
 main over

CyclicBarrier實現原理和使用場景

CyclicBarrier則是靠ReentrantLock lock和Condition trip屬性來實現同步
通俗解釋：CyclicBarrier需要阻塞全部執行緒到await狀態，然後全部執行緒再全部被喚醒執行。想像有一個欄杆攔住五隻羊，需要當五隻羊一起站在欄杆時，欄杆才會被拉起，此時所有的羊都可以飛跑出羊圈
使用示例

public static void main(String[] args) throws Exception {
    CyclicBarrier barrier = new CyclicBarrier(2);
    CompletableFuture.runAsync(()->{
        try {
            System.out.println("CompletableFuture run start-"+ Clock.systemUTC().millis());
            barrier.await(); //需要等待main執行緒也執行到await狀態才能繼續執行
            System.out.println("CompletableFuture run over-"+ Clock.systemUTC().millis());
        }catch (Exception e){
            throw new RuntimeException(e);
        }
    });
    Thread.sleep(1000);
    //和CompletableFuture執行緒相互等待
    barrier.await();
    System.out.println("main run over!");
}
---------------輸出結果------------------
CompletableFuture run start-1609822588881
main run over!
CompletableFuture run over-1609822589880

StampedLock

StampedLock不是藉助AQS，而是自己內部維護多個狀態值，並配合cas實現的
StampedLock具有三種模式：寫模式、讀模式、樂觀讀模式
StampedLock的讀寫鎖可以相互轉換

//獲取讀鎖，自旋獲取，返回一個戳值
public long readLock()
//嘗試加讀鎖，不成功返回0
public long tryReadLock()
//解鎖
public void unlockRead(long stamp) 
//獲取寫鎖，自旋獲取，返回一個戳值
public long writeLock()
//嘗試加寫鎖，不成功返回0
public long tryWriteLock()
//解鎖
public void unlockWrite(long stamp)
//嘗試樂觀讀讀取一個時間戳，並配合validate方法校驗時間戳的有效性
public long tryOptimisticRead()
//驗證stamp是否有效
public boolean validate(long stamp)

使用示例

public static void main(String[] args) throws Exception {
    StampedLock stampedLock = new StampedLock();
    long stamp = stampedLock.tryOptimisticRead();
    //判斷版本號是否生效
    if (!stampedLock.validate(stamp)) {
        //獲取讀鎖，會空轉
        stamp = stampedLock.readLock();
        long writeStamp = stampedLock.tryConvertToWriteLock(stamp);
        if (writeStamp != 0) { //成功轉為寫鎖
            //fixme 業務操作
            stampedLock.unlockWrite(writeStamp);
        } else {
            stampedLock.unlockRead(stamp);
            //嘗試獲取寫讀
            stamp = stampedLock.tryWriteLock();
            if (stamp != 0) {
                //fixme 業務操作
                stampedLock.unlockWrite(writeStamp);
            }
        }
    }
}

歡迎指正文中錯誤

參考文章

並發之Striped64（l累加器）

Tags: AQS cas JAVA JAVA基礎篇同步同步鎖面試

基礎篇：JAVA原子組件和同步組件

前言

關注公眾號，一起交流，微信搜一搜: 潛行前行

原子組件的實現原理CAS

應用場景

原子組件

基本類型原子類

引用類原子類

ABA問題

數組原子類

屬性原子更新類

累加器

同步組件的實現原理

同步組件

ReentrantLock、ReentrantReadWriteLock

Semaphore實現原理和使用場景

CountDownLatch實現原理和使用場景

CyclicBarrier實現原理和使用場景

StampedLock

歡迎指正文中錯誤

參考文章

VirMach 便宜 VPS

QNews

基礎篇：JAVA原子組件和同步組件

前言

關注公眾號，一起交流，微信搜一搜: 潛行前行

原子組件的實現原理CAS

應用場景

原子組件

基本類型原子類

引用類原子類

ABA問題

數組原子類

屬性原子更新類

累加器

同步組件的實現原理

同步組件

ReentrantLock、ReentrantReadWriteLock

Semaphore實現原理和使用場景

CountDownLatch實現原理和使用場景

CyclicBarrier實現原理和使用場景

StampedLock

歡迎指正文中錯誤

參考文章

分享此文：

Related Posts

S2-052 漏洞復現

配置manjaro源與系統更新

要飛4.7億公里：天問一號到火星的時間准了！

全國44城市地鐵排行：第三太猛！廣州都不如它

VirMach 便宜 VPS

QNews

熱門文章

熱門搜尋