基礎(chǔ)篇：Java原子組件和同步組件
作者：cscw
在使用多線(xiàn)程并發(fā)編程的時(shí)，經(jīng)常會(huì)遇到對(duì)共享變量修改操作。此時(shí)我們可以選擇ConcurrentHashMap，ConcurrentLinkedQueue來(lái)進(jìn)行安全地存儲(chǔ)數(shù)據(jù)。
本文轉(zhuǎn)載自微信公眾號(hào)「潛行前行」，作者cscw 。轉(zhuǎn)載本文請(qǐng)聯(lián)系潛行前行公眾號(hào)。
前言
在使用多線(xiàn)程并發(fā)編程的時(shí)，經(jīng)常會(huì)遇到對(duì)共享變量修改操作。此時(shí)我們可以選擇ConcurrentHashMap，ConcurrentLinkedQueue來(lái)進(jìn)行安全地存儲(chǔ)數(shù)據(jù)。但如果單單是涉及狀態(tài)的修改，線(xiàn)程執(zhí)行順序問(wèn)題，使用Atomic開(kāi)頭的原子組件或者ReentrantLock、CyclicBarrier之類(lèi)的同步組件，會(huì)是更好的選擇，下面將一一介紹它們的原理和用法
- 原子組件的實(shí)現(xiàn)原理CAS
- AtomicBoolean、AtomicIntegerArray等原子組件的用法、
- 同步組件的實(shí)現(xiàn)原理
- ReentrantLock、CyclicBarrier等同步組件的用法
原子組件的實(shí)現(xiàn)原理CAS
- cas的底層實(shí)現(xiàn)可以看下之前寫(xiě)的一篇文章:詳解鎖原理，synchronized、volatile+cas底層實(shí)現(xiàn)[1]
應(yīng)用場(chǎng)景
- 可用來(lái)實(shí)現(xiàn)變量、狀態(tài)在多線(xiàn)程下的原子性操作
- 可用于實(shí)現(xiàn)同步鎖(ReentrantLock)
原子組件
- 原子組件的原子性操作是靠使用cas來(lái)自旋操作volatile變量實(shí)現(xiàn)的
- volatile的類(lèi)型變量保證變量被修改時(shí)，其他線(xiàn)程都能看到最新的值
- cas則保證value的修改操作是原子性的，不會(huì)被中斷
基本類(lèi)型原子類(lèi)
- AtomicBoolean //布爾類(lèi)型
- AtomicInteger //正整型數(shù)類(lèi)型
- AtomicLong //長(zhǎ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());
- }
- ---------------輸出結(jié)果------------------
- boolean value is:false
引用類(lèi)原子類(lèi)
- AtomicReference
- //加時(shí)間戳版本的引用類(lèi)原子類(lèi)
- AtomicStampedReference
- //相當(dāng)于AtomicStampedReference，AtomicMarkableReference關(guān)心的是
- //變量是否還是原來(lái)變量，中間被修改過(guò)也無(wú)所謂
- AtomicMarkableReference
- AtomicReference的源碼如下，它內(nèi)部定義了一個(gè)volatile V value，并借助VarHandle(具體子類(lèi)是FieldInstanceReadWrite)實(shí)現(xiàn)原子操作，MethodHandles會(huì)幫忙計(jì)算value在類(lèi)的偏移位置，最后在VarHandle調(diào)用Unsafe.public final native boolean compareAndSetReference(Object o, long offset, Object expected, Object x)方法原子修改對(duì)象的屬性
- 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準(zhǔn)備將變量的值從A改為B，然而這期間線(xiàn)程Y將變量的值從A改為C，然后再改為A;最后線(xiàn)程X檢測(cè)變量值是A，并置換為B。但實(shí)際上，A已經(jīng)不再是原來(lái)的A了
- 解決方法，是把變量定為唯一類(lèi)型。值可以加上版本號(hào)，或者時(shí)間戳。如加上版本號(hào)，線(xiàn)程Y的修改變?yōu)锳1->B2->A3，此時(shí)線(xiàn)程X再更新則可以判斷出A1不等于A3
- AtomicStampedReference的實(shí)現(xiàn)和AtomicReference差不多，不過(guò)它原子修改的變量是volatile Pair
pair;，Pair是其內(nèi)部類(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)心當(dāng)前變量是否還是原先的變量，則可以使用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("對(duì)象："+reference.getReference().name+";版本號(hào)："+reference.getStamp());
- }
- }
- class Test{
- Test(String name){ this.name = name; }
- public String name;
- }
- ---------------輸出結(jié)果------------------
- 對(duì)象：world;版本號(hào)：2
數(shù)組原子類(lèi)
- AtomicIntegerArray //整型數(shù)組
- AtomicLongArray //長(zhǎng)整型數(shù)組
- AtomicReferenceArray //引用類(lèi)型數(shù)組
- 數(shù)組原子類(lèi)內(nèi)部會(huì)初始一個(gè)final的數(shù)組，它把整個(gè)數(shù)組當(dāng)做一個(gè)對(duì)象，然后根據(jù)下標(biāo)index計(jì)算法元素偏移量，再調(diào)用UNSAFE.compareAndSetReference進(jìn)行原子操作。數(shù)組并沒(méi)被volatile修飾，為了保證元素類(lèi)型在不同線(xiàn)程的可見(jiàn)，獲取元素使用到了UNSAFEpublic native Object getReferenceVolatile(Object o, long offset)方法來(lái)獲取實(shí)時(shí)的元素值
- 使用示例
- //元素默認(rèn)初始化為0
- AtomicIntegerArray array = new AtomicIntegerArray(2);
- // 下標(biāo)為０的元素，期待值是0，更新值是１
- array.compareAndSet(0,0,1);
- System.out.println(array.get(0));
- ---------------輸出結(jié)果------------------
- 1
屬性原子類(lèi)
- AtomicIntegerFieldUpdater
- AtomicLongFieldUpdater
- AtomicReferenceFieldUpdater
- 如果操作對(duì)象是某一類(lèi)型的屬性，可以使用AtomicIntegerFieldUpdater原子更新，不過(guò)類(lèi)的屬性需要定義成volatile修飾的變量，保證該屬性在各個(gè)線(xiàn)程的可見(jiàn)性，否則會(huì)報(bào)錯(cuò)
- 使用示例
- 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;
- }
- ---------------輸出結(jié)果------------------
- siting
- siting
累加器
- Striped64
- LongAccumulator
- LongAdder
- //accumulatorFunction：運(yùn)算規(guī)則，identity：初始值
- public LongAccumulator(LongBinaryOperator accumulatorFunction,long identity)
- LongAccumulator和LongAdder都繼承于Striped64，Striped64的主要思想是和ConcurrentHashMap有點(diǎn)類(lèi)似，分段計(jì)算，單個(gè)變量計(jì)算并發(fā)性能慢時(shí)，我們可以把數(shù)學(xué)運(yùn)算分散在多個(gè)變量，而需要計(jì)算總值時(shí)，再一一累加起來(lái)
- LongAdder相當(dāng)于LongAccumulator一個(gè)特例實(shí)現(xiàn)
- 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)程執(zhí)行完成，再獲取
- System.out.println(accumulator.get());
- }
- ---------------輸出結(jié)果------------------
- 100000
同步組件的實(shí)現(xiàn)原理
java的多數(shù)同步組件會(huì)在內(nèi)部維護(hù)一個(gè)狀態(tài)值，和原子組件一樣，修改狀態(tài)值時(shí)一般也是通過(guò)cas來(lái)實(shí)現(xiàn)。而狀態(tài)修改的維護(hù)工作被Doug Lea抽象出AbstractQueuedSynchronizer(AQS)來(lái)實(shí)現(xiàn)
AQS的原理可以看下之前寫(xiě)的一篇文章:詳解鎖原理，synchronized、volatile+cas底層實(shí)現(xiàn)[2]
同步組件
ReentrantLock、ReentrantReadWriteLock
- ReentrantLock、ReentrantReadWriteLock都是基于AQS(AbstractQueuedSynchronizer)實(shí)現(xiàn)的。因?yàn)樗鼈冇泄芥i和非公平鎖的區(qū)分，因此沒(méi)直接繼承AQS，而是使用內(nèi)部類(lèi)去繼承，公平鎖和非公平鎖各自實(shí)現(xiàn)AQS，ReentrantLock、ReentrantReadWriteLock再借助內(nèi)部類(lèi)來(lái)實(shí)現(xiàn)同步
- 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í)現(xiàn)原理和使用場(chǎng)景
- Semaphore和ReentrantLock一樣，也有公平和非公平競(jìng)爭(zhēng)鎖的策略，一樣也是通過(guò)內(nèi)部類(lèi)繼承AQS來(lái)實(shí)現(xiàn)同步
- 通俗解釋?zhuān)杭僭O(shè)有一口井，最多有三個(gè)人的位置打水。每有一個(gè)人打水，則需要占用一個(gè)位置。當(dāng)三個(gè)位置全部占滿(mǎn)時(shí)，第四個(gè)人需要打水，則要等待前三個(gè)人中一個(gè)離開(kāi)打水位，才能繼續(xù)獲取打水的位置
- 使用示例
- 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ú)阻塞結(jié)束 ");
- }else {
- System.out.println(Thread.currentThread().toString() + " 被阻塞結(jié)束 ");
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
- //保證CompletableFuture 線(xiàn)程被執(zhí)行，主線(xiàn)程再結(jié)束
- Thread.sleep(2000);
- }
- ---------------輸出結(jié)果------------------
- 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] 被阻塞結(jié)束
- Thread[ForkJoinPool.commonPool-worker-5,5,main] 無(wú)阻塞結(jié)束
- Thread[ForkJoinPool.commonPool-worker-19,5,main] 無(wú)阻塞結(jié)束
可以看出三個(gè)線(xiàn)程，因?yàn)樾盘?hào)量設(shè)定為2，第三個(gè)線(xiàn)程是無(wú)法獲取信息成功的，會(huì)打印阻塞結(jié)束
CountDownLatch實(shí)現(xiàn)原理和使用場(chǎng)景
- CountDownLatch也是靠AQS實(shí)現(xiàn)的同步操作
- 通俗解釋?zhuān)和嬗螒驎r(shí)，假如主線(xiàn)任務(wù)需要靠完成五個(gè)小任務(wù)，主線(xiàn)任務(wù)才能繼續(xù)進(jìn)行時(shí)。此時(shí)可以用CountDownLatch，主線(xiàn)任務(wù)阻塞等待，每完成一小任務(wù)，就done一次計(jì)數(shù)，直到五個(gè)小任務(wù)全部被執(zhí)行才能觸發(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 ");
- }
- ---------------輸出結(jié)果------------------
- CompletableFuture over
- CompletableFuture over
- main over
CyclicBarrier實(shí)現(xiàn)原理和使用場(chǎng)景
- CyclicBarrier則是靠ReentrantLock lock和Condition trip屬性來(lái)實(shí)現(xiàn)同步
- 通俗解釋?zhuān)篊yclicBarrier需要阻塞全部線(xiàn)程到await狀態(tài)，然后全部線(xiàn)程再全部被喚醒執(zhí)行。想象有一個(gè)欄桿攔住五只羊，需要當(dāng)五只羊一起站在欄桿時(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)程也執(zhí)行到await狀態(tài)才能繼續(xù)執(zhí)行
- 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!");
- }
- ---------------輸出結(jié)果------------------
- CompletableFuture run start-1609822588881
- main run over!
- CompletableFuture run over-1609822589880
StampedLock
- StampedLock不是借助AQS，而是自己內(nèi)部維護(hù)多個(gè)狀態(tài)值，并配合cas實(shí)現(xiàn)的
- StampedLock具有三種模式：寫(xiě)模式、讀模式、樂(lè)觀讀模式
- StampedLock的讀寫(xiě)鎖可以相互轉(zhuǎn)換
- //獲取讀鎖，自旋獲取，返回一個(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è)觀讀讀取一個(gè)時(shí)間戳，并配合validate方法校驗(yàn)時(shí)間戳的有效性
- public long tryOptimisticRead()
- //驗(yàn)證stamp是否有效
- public boolean validate(long stamp)
- 使用示例
- public static void main(String[] args) throws Exception {
- StampedLock stampedLock = new StampedLock();
- long stamp = stampedLock.tryOptimisticRead();
- //判斷版本號(hào)是否生效
- if (!stampedLock.validate(stamp)) {
- //獲取讀鎖，會(huì)空轉(zhuǎn)
- stamp = stampedLock.readLock();
- long writeStamp = stampedLock.tryConvertToWriteLock(stamp);
- if (writeStamp != 0) { //成功轉(zhuǎn)為寫(xiě)鎖
- //fixme 業(yè)務(wù)操作
- stampedLock.unlockWrite(writeStamp);
- } else {
- stampedLock.unlockRead(stamp);
- //嘗試獲取寫(xiě)讀
- stamp = stampedLock.tryWriteLock();
- if (stamp != 0) {
- //fixme 業(yè)務(wù)操作
- stampedLock.unlockWrite(writeStamp);
- }
- }
- }
- }
參考文章
- 并發(fā)之Striped64(l累加器)[3]
參考資料
[1]詳解鎖原理，synchronized、volatile+cas底層實(shí)現(xiàn): https://juejin.cn/post/6854573210768900110
[2]詳解鎖原理，synchronized、volatile+cas底層實(shí)現(xiàn): https://juejin.cn/post/6854573210768900110
[3]并發(fā)之Striped64(l累加器): https://www.cnblogs.com/gosaint/p/9129867.html
責(zé)任編輯：武曉燕
來(lái)源：
潛行前行
