Java中如何使用原子組件和同步組件

Java中如何使用原子組件和同步組件，針對這個(gè)問(wèn)題，這篇文章詳細介紹了相對應的分析和解答，希望可以幫助更多想解決這個(gè)問(wèn)題的小伙伴找到更簡(jiǎn)單易行的方法。

原子組件的實(shí)現原理CAS

• cas的底層實(shí)現可以看下之前寫(xiě)的一篇文章:詳解鎖原理，synchronized、volatile+cas底層實(shí)現[1]

應用場(chǎng)景

• 可用來(lái)實(shí)現變量、狀態(tài)在多線(xiàn)程下的原子性操作

• 可用于實(shí)現同步鎖(ReentrantLock)

原子組件

• 原子組件的原子性操作是靠使用cas來(lái)自旋操作volatile變量實(shí)現的

• volatile的類(lèi)型變量保證變量被修改時(shí)，其他線(xiàn)程都能看到最新的值

• cas則保證value的修改操作是原子性的，不會(huì )被中斷

基本類(lèi)型原子類(lèi)

AtomicBoolean //布爾類(lèi)型 AtomicInteger //正整型數類(lèi)型 AtomicLong   //長(cháng)整型類(lèi)型

使用示例

public static void main(String[] args) throws Exception {     AtomicBoolean atomicBoolean = new AtomicBoolean(false);     //異步線(xiàn)程修改atomicBoolean     CompletableFuture<Void> future = CompletableFuture.runAsync(() ->{         try {             Thread.sleep(1000); //保證異步線(xiàn)程是在主線(xiàn)程之后修改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

引用類(lèi)原子類(lèi)

AtomicReference //加時(shí)間戳版本的引用類(lèi)原子類(lèi) AtomicStampedReference //相當于A(yíng)tomicStampedReference，AtomicMarkableReference關(guān)心的是 //變量是否還是原來(lái)變量，中間被修改過(guò)也無(wú)所謂 AtomicMarkableReference

• AtomicReference的源碼如下，它內部定義了一個(gè)volatile V  value，并借助VarHandle(具體子類(lèi)是FieldInstanceReadWrite)實(shí)現原子操作，MethodHandles會(huì )幫忙計算value在類(lèi)的偏移位置，最后在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問(wèn)題

• 線(xiàn)程X準備將變量的值從A改為B，然而這期間線(xiàn)程Y將變量的值從A改為C，然后再改為A;最后線(xiàn)程X檢測變量值是A，并置換為B。但實(shí)際上，A已經(jīng)不再是原來(lái)的A了

• 解決方法，是把變量定為唯一類(lèi)型。值可以加上版本號，或者時(shí)間戳。如加上版本號，線(xiàn)程Y的修改變?yōu)锳1->B2->A3，此時(shí)線(xiàn)程X再更新則可以判斷出A1不等于A(yíng)3

• AtomicStampedReference的實(shí)現和AtomicReference差不多，不過(guò)它原子修改的變量是volatile Pair

  pair;，Pair是其內部類(lèi)。AtomicStampedReference可以用來(lái)解決ABA問(wèn)題

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;

• 如果我們不關(guān)心變量在中間過(guò)程是否被修改過(guò)，而只是關(guān)心當前變量是否還是原先的變量，則可以使用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

數組原子類(lèi)

AtomicIntegerArray //整型數組  AtomicLongArray //長(cháng)整型數組  AtomicReferenceArray //引用類(lèi)型數組

• 數組原子類(lèi)內部會(huì )初始一個(gè)final的數組，它把整個(gè)數組當做一個(gè)對象，然后根據下標index計算法元素偏移量，再調用UNSAFE.compareAndSetReference進(jìn)行原子操作。數組并沒(méi)被volatile修飾，為了保證元素類(lèi)型在不同線(xiàn)程的可見(jiàn)，獲取元素使用到了UNSAFEpublic  native Object getReferenceVolatile(Object o, long offset)方法來(lái)獲取實(shí)時(shí)的元素值

• 使用示例

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

屬性原子類(lèi)

AtomicIntegerFieldUpdater  AtomicLongFieldUpdater AtomicReferenceFieldUpdater

• 如果操作對象是某一類(lèi)型的屬性，可以使用AtomicIntegerFieldUpdater原子更新，不過(guò)類(lèi)的屬性需要定義成volatile修飾的變量，保證該屬性在各個(gè)線(xiàn)程的可見(jiàn)性，否則會(huì )報錯

• 使用示例

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有點(diǎn)類(lèi)似，分段計算，單個(gè)變量計算并發(fā)性能慢時(shí)，我們可以把數學(xué)運算分散在多個(gè)變量，而需要計算總值時(shí)，再一一累加起來(lái)

• LongAdder相當于LongAccumulator一個(gè)特例實(shí)現

• 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線(xiàn)程執行完成，再獲取     System.out.println(accumulator.get()); } ---------------輸出結果------------------ 100000

同步組件的實(shí)現原理

java的多數同步組件會(huì )在內部維護一個(gè)狀態(tài)值，和原子組件一樣，修改狀態(tài)值時(shí)一般也是通過(guò)cas來(lái)實(shí)現。而狀態(tài)修改的維護工作被Doug  Lea抽象出AbstractQueuedSynchronizer(AQS)來(lái)實(shí)現

AQS的原理可以看下之前寫(xiě)的一篇文章:詳解鎖原理，synchronized、volatile+cas底層實(shí)現[2]

同步組件

ReentrantLock、ReentrantReadWriteLock

• ReentrantLock、ReentrantReadWriteLock都是基于A(yíng)QS(AbstractQueuedSynchronizer)實(shí)現的。因為它們有公平鎖和非公平鎖的區分，因此沒(méi)直接繼承AQS，而是使用內部類(lèi)去繼承，公平鎖和非公平鎖各自實(shí)現AQS，ReentrantLock、ReentrantReadWriteLock再借助內部類(lèi)來(lái)實(shí)現同步

• ReentrantLock的使用示例

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

• ReentrantReadWriteLock的使用示例

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

Semaphore實(shí)現原理和使用場(chǎng)景

• Semaphore和ReentrantLock一樣，也有公平和非公平競爭鎖的策略，一樣也是通過(guò)內部類(lèi)繼承AQS來(lái)實(shí)現同步

• 通俗解釋?zhuān)杭僭O有一口井，最多有三個(gè)人的位置打水。每有一個(gè)人打水，則需要占用一個(gè)位置。當三個(gè)位置全部占滿(mǎn)時(shí)，第四個(gè)人需要打水，則要等待前三個(gè)人中一個(gè)離開(kāi)打水位，才能繼續獲取打水的位置

• 使用示例

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() + " 無(wú)阻塞結束 ");                 }else {                     System.out.println(Thread.currentThread().toString() + " 被阻塞結束 ");                 }             } catch (Exception e) {                 throw new RuntimeException(e);             }         });     //保證CompletableFuture 線(xiàn)程被執行，主線(xiàn)程再結束     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] 無(wú)阻塞結束  Thread[ForkJoinPool.commonPool-worker-19,5,main] 無(wú)阻塞結束

可以看出三個(gè)線(xiàn)程，因為信號量設定為2，第三個(gè)線(xiàn)程是無(wú)法獲取信息成功的，會(huì )打印阻塞結束

CountDownLatch實(shí)現原理和使用場(chǎng)景

• CountDownLatch也是靠AQS實(shí)現的同步操作

• 通俗解釋?zhuān)和嬗螒驎r(shí)，假如主線(xiàn)任務(wù)需要靠完成五個(gè)小任務(wù)，主線(xiàn)任務(wù)才能繼續進(jìn)行時(shí)。此時(shí)可以用CountDownLatch，主線(xiàn)任務(wù)阻塞等待，每完成一小任務(wù)，就done一次計數，直到五個(gè)小任務(wù)全部被執行才能觸發(fā)主線(xiàn)

• 使用示例

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線(xiàn)程的完成     count.await();     System.out.println(" main over "); } ---------------輸出結果------------------  CompletableFuture over   CompletableFuture over   main over

CyclicBarrier實(shí)現原理和使用場(chǎng)景

• CyclicBarrier則是靠ReentrantLock lock和Condition trip屬性來(lái)實(shí)現同步

• 通俗解釋?zhuān)篊yclicBarrier需要阻塞全部線(xiàn)程到await狀態(tài)，然后全部線(xiàn)程再全部被喚醒執行。想象有一個(gè)欄桿攔住五只羊，需要當五只羊一起站在欄桿時(shí)，欄桿才會(huì )被拉起，此時(shí)所有的羊都可以飛跑出羊圈

• 使用示例

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線(xiàn)程也執行到await狀態(tài)才能繼續執行             System.out.println("CompletableFuture run over-"+ Clock.systemUTC().millis());         }catch (Exception e){             throw new RuntimeException(e);         }     });     Thread.sleep(1000);     //和CompletableFuture線(xiàn)程相互等待     barrier.await();     System.out.println("main run over!"); } ---------------輸出結果------------------ CompletableFuture run start-1609822588881 main run over! CompletableFuture run over-1609822589880

StampedLock

• StampedLock不是借助AQS，而是自己內部維護多個(gè)狀態(tài)值，并配合cas實(shí)現的

• StampedLock具有三種模式：寫(xiě)模式、讀模式、樂(lè )觀(guān)讀模式

• StampedLock的讀寫(xiě)鎖可以相互轉換

//獲取讀鎖，自旋獲取，返回一個(gè)戳值 public long readLock() //嘗試加讀鎖，不成功返回0 public long tryReadLock() //解鎖 public void unlockRead(long stamp)  //獲取寫(xiě)鎖，自旋獲取，返回一個(gè)戳值 public long writeLock() //嘗試加寫(xiě)鎖，不成功返回0 public long tryWriteLock() //解鎖 public void unlockWrite(long stamp) //嘗試樂(lè )觀(guān)讀讀取一個(gè)時(shí)間戳，并配合validate方法校驗時(shí)間戳的有效性 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)) {         //獲取讀鎖，會(huì )空轉         stamp = stampedLock.readLock();         long writeStamp = stampedLock.tryConvertToWriteLock(stamp);         if (writeStamp != 0) { //成功轉為寫(xiě)鎖             //fixme 業(yè)務(wù)操作             stampedLock.unlockWrite(writeStamp);         } else {             stampedLock.unlockRead(stamp);             //嘗試獲取寫(xiě)讀             stamp = stampedLock.tryWriteLock();             if (stamp != 0) {                 //fixme 業(yè)務(wù)操作                 stampedLock.unlockWrite(writeStamp);             }         }     } }