思考：

• ReentrantLock與Synchronized的區(qū)別羡鸥？面對(duì)增加屬性如：公平鎖是如何實(shí)現(xiàn)的呢咬摇？
• 讀寫(xiě)鎖是共享-互斥鎖儡陨，讀鎖是如何多線程共享辙诞，寫(xiě)鎖是如何做到互斥辙售？
• wait/nofity會(huì)導(dǎo)致死鎖嗎，使用它們先喚醒再掛起而導(dǎo)致死鎖飞涂，park為何不會(huì)呢旦部？

AQS

• 如果需要了解lock鎖祈搜，繞不開(kāi)AQS（abstract Queue Synchronized 抽象隊(duì)列同步器）其定義了一套多線程訪問(wèn)共享資源的同步器框架
• AQS維護(hù)了維護(hù)了一個(gè) volatile int state（代表共享資源）和一個(gè)FIFO線程等待隊(duì)列（多線程爭(zhēng)用資源被阻塞時(shí)會(huì)進(jìn)入此隊(duì)列）

Node

• 隊(duì)列中節(jié)點(diǎn)由內(nèi)部類(lèi)Node位組成的隊(duì)列“CLH”（三個(gè)人名：Craig, Landin 和 Hagersten），CLH隊(duì)列本質(zhì)上就是一個(gè)雙向鏈表Node就是該鏈表的節(jié)點(diǎn)士八，數(shù)據(jù)如下

  
  
  node.png

1. volatile int waitStatus :當(dāng)前節(jié)點(diǎn)Node的等待狀態(tài)標(biāo)志位容燕，標(biāo)記該節(jié)點(diǎn)當(dāng)前情況下處于何種狀態(tài)

• AQS源碼中大量運(yùn)用waitStatus判斷節(jié)點(diǎn)狀態(tài)：

2. nextWaiter:保存條件隊(duì)列

• AQS中阻塞隊(duì)列采用的是用雙向鏈表保存忍坷，用prve和next相互鏈接粘舟。而AQS中條件隊(duì)列是使用單向列表保存的，用 nextWaiter來(lái)連接佩研。阻塞隊(duì)列和條件隊(duì)列并不是使用的相同的數(shù)據(jù)結(jié)構(gòu)柑肴。

// 共享模式
static final Node SHARED = new Node();
// 獨(dú)占模式
static final Node EXCLUSIVE = null;
// 其他模式
// 其他非空值：條件等待節(jié)點(diǎn)（調(diào)用Condition的await方法的時(shí)候）

• nextWaiter實(shí)際上標(biāo)記的就是在該節(jié)點(diǎn)喚醒后依據(jù)該節(jié)點(diǎn)的狀態(tài)判斷是否依據(jù)條件喚醒下一個(gè)節(jié)點(diǎn)

ReentrantLock

• RentrantLock比較熟悉旬薯，這里分成 加鎖晰骑，入隊(duì)，阻塞绊序，喚醒硕舆，出隊(duì)，解鎖六個(gè)步驟來(lái)分析

加鎖

1. 使用公平鎖及非公平鎖

public ReentrantLock(boolean fair) {
    this.sync = (ReentrantLock.Sync)(fair ? new ReentrantLock.FairSync() : new ReentrantLock.NonfairSync());
}

2. 使用lock 調(diào)用 AQS中的acquire(1)

public final void acquire(int arg) {
    //AQS實(shí)現(xiàn)類(lèi)
    if (!this.tryAcquire(arg) && this.acquireQueued(this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), arg)) {
        selfInterrupt();
    }

}

//tryAcquire（1） 中分成公平鎖和非公平鎖實(shí)現(xiàn)
final boolean nonfairTryAcquire(int acquires) {

    Thread current = Thread.currentThread();
    //獲取當(dāng)前state狀態(tài)
    int c = this.getState();
    if (c == 0) { //當(dāng)前狀態(tài)為0表示可加鎖用
        //如果state  CAS操作成功骤公，將state替換為1并標(biāo)記當(dāng)前線程到AQS隊(duì)列中正在運(yùn)行的
        if (this.compareAndSetState(0, acquires)) {  
        /**
        * 公平鎖與非公平鎖唯一的區(qū)別即為先判斷 !this.hasQueuedPredecessors() 阻塞隊(duì)列中是否存在數(shù)據(jù)
        */
        //if (!this.hasQueuedPredecessors() && this.compareAndSetState(0, acquires)) {
            this.setExclusiveOwnerThread(current);
            return true;
        }
    } else if (current == this.getExclusiveOwnerThread()) {
    //可重入鎖設(shè)置
        int nextc = c + acquires;
        if (nextc < 0) {
            throw new Error("Maximum lock count exceeded");
        }

        this.setState(nextc); //state++ ; 
        return true;
    }

    return false;
}

//公平鎖判斷阻塞隊(duì)列是否存在元素
 public final boolean hasQueuedPredecessors() {
        AbstractQueuedSynchronizer.Node h;
        if ((h = this.head) != null) { //隊(duì)頭表示當(dāng)前運(yùn)行線程
            AbstractQueuedSynchronizer.Node s;
            if ((s = h.next) == null || s.waitStatus > 0) {  //隊(duì)頭的下一個(gè)節(jié)點(diǎn)失效抚官，對(duì)其致空
                s = null;
        //從隊(duì)尾循環(huán)獲取對(duì)頭后面第一個(gè)waitStatus等待節(jié)點(diǎn)
                for(AbstractQueuedSynchronizer.Node p = this.tail; p != h && p != null; p = p.prev) {
                    if (p.waitStatus <= 0) {
                        s = p;
                    }
                }
            }
      //表示有正常線程在隊(duì)列中等待數(shù)據(jù)，且不是重入的阶捆，這時(shí)tryAcquire()運(yùn)行完畢=> 失敗凌节，運(yùn)行addWaiter()添加節(jié)點(diǎn)到阻塞隊(duì)列
            if (s != null && s.thread != Thread.currentThread()) {
                return true;
            }
        }

        return false;
    }

• 通過(guò)以上源碼分析可知：

  1. 當(dāng)Lock.lock()時(shí)會(huì)調(diào)用AQS的acquire(1)根據(jù)是否公平/非公平調(diào)用不同的實(shí)現(xiàn)tryAcquire
     • 如果公平鎖會(huì)首先判斷AQS阻塞隊(duì)列中是否有元素存在，如果有等待線程則直接加入隊(duì)尾
  2. 如果當(dāng)前state狀態(tài)為0趁猴，代表鎖可用刊咳，通過(guò)CAS替換為1，成功線程獲取鎖儡司，設(shè)置AQS運(yùn)行線程為當(dāng)前線程
  3. 若狀態(tài)不為0，查看是否線程一致余指，若一致表示可重入鎖捕犬，只需增加state++即可

  4. 1~3表示lock加鎖成功跷坝，正常進(jìn)入運(yùn)行，否則嘗試鎖失敗碉碉，運(yùn)行acquireQueued（）入隊(duì)操作

     
     
     image.png

• 調(diào)用關(guān)系圖： 有關(guān)于status狀態(tài)修改柴钻，生成節(jié)點(diǎn)，入隊(duì)垢粮，出隊(duì)等操作均有AQS完成

  
  
  image.png

入隊(duì)

• 以上加鎖失敗后在AQS中入隊(duì)操作addWaiter()

private AbstractQueuedSynchronizer.Node addWaiter(AbstractQueuedSynchronizer.Node mode) {
        AbstractQueuedSynchronizer.Node node = new AbstractQueuedSynchronizer.Node(mode);

        AbstractQueuedSynchronizer.Node oldTail;
        do {
            while(true) { //首次進(jìn)入head,tail = null
                oldTail = this.tail;
                if (oldTail != null) {
                    node.setPrevRelaxed(oldTail);
                    break;
                }
        //jdk9之前是enq(node)贴届，換湯不換藥啦
                this.initializeSyncQueue();
            }
        } while(!this.compareAndSetTail(oldTail, node));

        oldTail.next = node;
        return node;
    }
    
 //生成兩個(gè)節(jié)點(diǎn)，head節(jié)點(diǎn)位Node() 蜡吧，head.next 才是真正的第一個(gè)入隊(duì)節(jié)點(diǎn)數(shù)據(jù)
 private final void initializeSyncQueue() {
    AbstractQueuedSynchronizer.Node h;
       if (HEAD.compareAndSet(this, (Void)null, h = new AbstractQueuedSynchronizer.Node())) {
          this.tail = h;
      }
}

1. 首次進(jìn)入head,tail進(jìn)入initializeSyncQueue() 初始化head為一個(gè)Node() ==> thread = null

2. while循環(huán)第2次 將node節(jié)點(diǎn)添加到tail后面即head.next = node 并CAS交換tail指向node

   
   
   image.png

阻塞

• 在AQS的acquireQueued()中實(shí)現(xiàn)

final boolean acquireQueued(AbstractQueuedSynchronizer.Node node, int arg) {
        boolean interrupted = false;

        try {
            while(true) {
                AbstractQueuedSynchronizer.Node p = node.predecessor(); //當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)
                if (p == this.head && this.tryAcquire(arg)) { //如果前驅(qū)節(jié)點(diǎn)為head,則嘗試加鎖
                    this.setHead(node); //加鎖成功毫蚓，設(shè)置head為當(dāng)前節(jié)點(diǎn)，同時(shí)設(shè)置node.thread = null, node.pre =null
                    p.next = null; //方便GC回收
                    return interrupted;
                }
        //獲取鎖失敗
                if (shouldParkAfterFailedAcquire(p, node)) {
                    interrupted |= this.parkAndCheckInterrupt();
                }
            }
        } catch (Throwable var5) {
            this.cancelAcquire(node);
            if (interrupted) {
                selfInterrupt();
            }

            throw var5;
        }
    }
    
    //移動(dòng)head到當(dāng)前Node
   private void setHead(AbstractQueuedSynchronizer.Node node) {
        this.head = node;
        node.thread = null;
        node.prev = null;
    }
    
 // 該方法主要靠前驅(qū)節(jié)點(diǎn)判斷當(dāng)前線程是否應(yīng)該被阻塞
private static boolean shouldParkAfterFailedAcquire(AbstractQueuedSynchronizer.Node pred, AbstractQueuedSynchronizer.Node node) {
        int ws = pred.waitStatus; //前驅(qū)節(jié)點(diǎn)狀態(tài)
        if (ws == -1) { //SIGNAL
            return true;
        } else {
            if (ws > 0) { //循環(huán)判斷當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)昔善，移除cancel節(jié)點(diǎn)
                do {
                  /**
                  * 循環(huán)查找取消節(jié)點(diǎn)的前任節(jié)點(diǎn)元潘，
                    * 直到找到不是取消狀態(tài)的節(jié)點(diǎn)，然后剔除是取消狀態(tài)的節(jié)點(diǎn)君仆，
                    * 關(guān)聯(lián)前任節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)為當(dāng)前節(jié)點(diǎn)
                  */
                    node.prev = pred = pred.prev;
                } while(pred.waitStatus > 0);

                pred.next = node; //將當(dāng)前節(jié)點(diǎn)添加到前驅(qū)的后置節(jié)點(diǎn)
            } else {
              /*
                * CAS設(shè)置前任節(jié)點(diǎn)等待狀態(tài)為SIGNAL翩概，
                * 設(shè)置成功表示當(dāng)前節(jié)點(diǎn)應(yīng)該被阻塞，下一次循環(huán)調(diào)用就會(huì)
                *  return  true
               */
                pred.compareAndSetWaitStatus(ws, -1); 
            }

            return false;
        }
    }
    
 /**
  * 把當(dāng)前線程掛起返咱，從而阻塞住線程的調(diào)用棧钥庇，同時(shí)返回當(dāng)前線程的中斷狀態(tài)。
 * 其內(nèi)部則是調(diào)用LockSupport工具類(lèi)的park()方法來(lái)阻塞該方法
 */
 private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
 }

image.png

1. 判斷當(dāng)前Node的前驅(qū)節(jié)點(diǎn)是否為head咖摹，若是則嘗試獲取鎖评姨，成功則將head 指向Node，并斷開(kāi)于前節(jié)點(diǎn)聯(lián)系楞艾，待GC回收前節(jié)點(diǎn)
2. 如果失敗参咙，則判斷前置節(jié)點(diǎn)是否為SIGNAL狀態(tài)，如果是硫眯，可以放心的park了蕴侧，因?yàn)闀?huì)被叫醒
3. 否則，循環(huán)查找前置節(jié)點(diǎn)是否為CANCEL狀態(tài)两入，如果是净宵，則移除，直到找到一個(gè)為SIGNAL或者默認(rèn)0裹纳，CAS替換為 -1择葡，待下次while循環(huán)返回true進(jìn)入阻塞
4. 阻塞函數(shù)調(diào)用LockSupport.park(this)線程被掛起了 ， Linux下的park查詢(xún)是否有許可剃氧，unpark函數(shù)為線程提供“許可(permit)”敏储，線程調(diào)用park函數(shù)則等待“許可”，這個(gè)“許可”是不能疊加的朋鞍，“許可”是一次性的已添。

解鎖 And 喚醒

• 當(dāng)調(diào)用lock.unlock解鎖后妥箕，如果status為0，則會(huì)調(diào)用unpark喚醒head后面第一個(gè)可用節(jié)點(diǎn)：

public final boolean release(int arg) {
        if (this.tryRelease(arg)) {
            AbstractQueuedSynchronizer.Node h = this.head;
            if (h != null && h.waitStatus != 0) { //全部解鎖成功進(jìn)入更舞，head不為null,且waitStatus 不為默認(rèn)值
                this.unparkSuccessor(h);
            }

            return true;
        } else {
            return false;
        }
    }
    
 @ReservedStackAccess
 protected final boolean tryRelease(int releases) {
      int c = this.getState() - releases;
      if (Thread.currentThread() != this.getExclusiveOwnerThread()) {
           throw new IllegalMonitorStateException();
      } else {
           boolean free = false;
           if (c == 0) { //可重入鎖全部解鎖成功
              free = true;
              this.setExclusiveOwnerThread((Thread)null); //設(shè)置AQS中Thread = null
           }

           this.setState(c);
           return free;
        }
  }
  
//
private void unparkSuccessor(AbstractQueuedSynchronizer.Node node) {
        int ws = node.waitStatus;
        if (ws < 0) {  //ws狀態(tài)可用畦幢，恢復(fù)默認(rèn)0
            node.compareAndSetWaitStatus(ws, 0);
        }

        AbstractQueuedSynchronizer.Node s = node.next; //獲取head下一個(gè)節(jié)點(diǎn)
        if (s == null || s.waitStatus > 0) { //從tail尾部遍歷獲取head下一個(gè)可用的節(jié)點(diǎn)s
            s = null;

            for(AbstractQueuedSynchronizer.Node p = this.tail; p != node && p != null; p = p.prev) {
                if (p.waitStatus <= 0) {
                    s = p;
                }
            }
        }

        if (s != null) { //存在為signal狀態(tài)的waitStatus喚醒它
            LockSupport.unpark(s.thread);
        }

}

image.png

• 注意，此時(shí)并沒(méi)有出隊(duì)操作哦缆蝉！那出隊(duì)是什么時(shí)候呢宇葱？
  • 就是被喚醒的線程獲取鎖的操作中出隊(duì)的；

出隊(duì)

• 在acquireQueued的while循環(huán)中刊头，即setHead移動(dòng)head操作黍瞧，回收原有的oldHead即出隊(duì)

ReentrantReadWriteLock

• 讀寫(xiě)鎖：寫(xiě)鎖和讀鎖是互斥的（這里的互斥是指線程間的互斥，當(dāng)前線程可以獲取到寫(xiě)鎖又獲取到讀鎖芽偏，但是獲取到了讀鎖不能繼續(xù)獲取寫(xiě)鎖）雷逆，這是因?yàn)樽x寫(xiě)鎖要保持寫(xiě)操作的可見(jiàn)性，如果允許讀鎖在被獲取的情況下對(duì)寫(xiě)鎖的獲取污尉，那么正在運(yùn)行的其他讀線程無(wú)法感知到當(dāng)前寫(xiě)線程的操作膀哲。因此，只有等待其他線程都釋放了讀鎖被碗，寫(xiě)鎖才能被當(dāng)前線程獲取某宪，而一旦寫(xiě)鎖被獲取，其他讀寫(xiě)線程的后續(xù)訪問(wèn)都會(huì)被阻塞锐朴。
• 注意讀寫(xiě)鎖中的幾個(gè)常量值及位運(yùn)算：
  • 分析可知：讀寫(xiě)鎖同樣適用AQS的int型status標(biāo)記兴喂，高16位表示讀鎖次數(shù)，低16位表示寫(xiě)鎖次數(shù)焚志，那么為何使用一個(gè)值而不是分開(kāi)使用兩個(gè)呢衣迷？

static final int SHARED_SHIFT = 16; //移位個(gè)數(shù)，讀鎖或者寫(xiě)鎖通過(guò)移位確定
static final int SHARED_UNIT = 65536;  //0X10000
static final int MAX_COUNT = 65535; //0Xffff
static final int EXCLUSIVE_MASK = 65535; //0Xffff

//獲取分享個(gè)數(shù)：即讀鎖次數(shù)
static int sharedCount(int c) {
    return c >>> 16;
}

//獨(dú)占鎖獲取鎖次數(shù)： 用于讀鎖操作 c為AQS中的status
static int exclusiveCount(int c) {
      return c & '\uffff'; 
}

讀寫(xiě)鎖加鎖

• 以下為讀寫(xiě)鎖加鎖示例圖,注意配合源碼查看分析酱酬，這里篇幅原因就不貼源碼了

  
  
  image.png

讀寫(xiě)鎖解鎖

• 讀鎖壶谒，寫(xiě)鎖的解鎖操作示意圖如下

  
  
  image.png

• 知道你可能對(duì)流程圖一知半解，舉個(gè)稍微復(fù)雜的例子來(lái)說(shuō)明思路：對(duì)于整個(gè)過(guò)程圖膳沽，當(dāng)寫(xiě)鎖獲取后的排隊(duì)情況為寫(xiě)汗菜，讀，讀挑社，寫(xiě)陨界，讀流程圖如下；

  
  
  image.png

思考

1. park函數(shù)為何不會(huì)導(dǎo)致死鎖呢痛阻？

//park調(diào)用鏈 LockSupport.park() ->unsafe.park() -> posixThread.park()
void Parker::park(bool isAbsolute, jlong time) {
  if (_counter > 0) {
       //已經(jīng)有許可了菌瘪，用掉當(dāng)前許可
      _counter = 0 ;
     //使用內(nèi)存屏障，確保 _counter賦值為0(寫(xiě)入操作)能夠被內(nèi)存屏障之后的讀操作獲取內(nèi)存屏障事前的結(jié)果阱当，也就是能夠正確的讀到0
      OrderAccess::fence();
     //立即返回
      return ;
  }

//這里表示線程互斥變量鎖成功了
  int status ;
  if (_counter > 0)  {
    // 有許可了麻车，返回
    _counter = 0;
    //對(duì)互斥變量解鎖
    status = pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant") ;
    OrderAccess::fence();
    return;
  }
  ...省略代碼...
  
  //沒(méi)有許可缀皱，調(diào)用wait等待
  if (time == 0) {
    //把調(diào)用線程放到等待條件的線程列表上斗这，然后對(duì)互斥變量解鎖动猬，（這兩是原子操作），這個(gè)時(shí)候線程進(jìn)入等待表箭，當(dāng)它返回時(shí)赁咙，互斥變量再次被鎖住。
  //成功返回0免钻，否則返回錯(cuò)誤編號(hào)
    status = pthread_cond_wait (_cond, _mutex) ;
  } else {
  //同pthread_cond_wait彼水，只是多了一個(gè)超時(shí)，如果超時(shí)還沒(méi)有條件出現(xiàn)极舔，那么重新獲取胡吃?xún)扇缓蠓祷劐e(cuò)誤碼 ETIMEDOUT
    status = os::Linux::safe_cond_timedwait (_cond, _mutex, &absTime) ;
    }
  }

#ifdef ASSERT
  pthread_sigmask(SIG_SETMASK, &oldsigs, NULL);
#endif
 //等待結(jié)束后凤覆，許可被消耗，改為0  _counter = 0 ;
//釋放互斥量的鎖
  status = pthread_mutex_unlock(_mutex) ;
  assert_status(status == 0, status, "invariant") ;
  // If externally suspended while waiting, re-suspend 
    if (jt->handle_special_suspend_equivalent_condition()) {
    jt->java_suspend_self();
  }
//加入內(nèi)存屏障指令
  OrderAccess::fence();
}

void Parker::unpark() {
  int s, status ;
 //給互斥量加鎖拆魏，如果互斥量已經(jīng)上鎖盯桦，則阻塞到互斥量被解鎖
//park進(jìn)入wait時(shí)，_mutex會(huì)被釋放
  status = pthread_mutex_lock(_mutex);
  assert (status == 0, "invariant") ; 
  //存儲(chǔ)舊的_counter
  s = _counter; 
//許可改為1渤刃，每次調(diào)用都設(shè)置成發(fā)放許可
  _counter = 1;
  if (s < 1) {
     //之前沒(méi)有許可
      //默認(rèn)執(zhí)行 ,釋放信號(hào)拥峦，表明條件已經(jīng)滿(mǎn)足，將喚醒等待的線程
      status = pthread_cond_signal (_cond) ;
      //釋放鎖
      status = pthread_mutex_unlock(_mutex);
  
  } else {
   //一直有許可卖子，釋放掉自己加的鎖,有許可park本身就返回了
    pthread_mutex_unlock(_mutex);
  }
}

• LockSupport類(lèi)為線程阻塞喚醒提供了基礎(chǔ)略号，同時(shí)，在競(jìng)爭(zhēng)條件問(wèn)題上具有wait和notify無(wú)可比擬的優(yōu)勢(shì)洋闽。使用wait和notify組合時(shí)玄柠，某一線程在被另一線程notify之前必須要保證此線程已經(jīng)執(zhí)行到wait等待點(diǎn)，錯(cuò)過(guò)notify則可能永遠(yuǎn)都在等待诫舅，另外notify也不能保證喚醒指定的某線程羽利。反觀LockSupport，由于park與unpark引入了許可機(jī)制骚勘，許可邏輯為：
• park將許可在等于0的時(shí)候阻塞铐伴，等于1的時(shí)候返回并將許可減為0。
• unpark嘗試喚醒線程俏讹，許可設(shè)置成 1当宴。
• 根據(jù)這兩個(gè)邏輯，對(duì)于同一條線程泽疆，park與unpark先后操作的順序并不影響程序正確地執(zhí)行户矢。假如先執(zhí)行unpark操作，許可則為1殉疼，之后再執(zhí)行park操作梯浪，此時(shí)因?yàn)樵S可等于1直接返回往下執(zhí)行捌年，并不執(zhí)行阻塞操作。 最后挂洛，LockSupport的park與unpark組合真正解耦了線程之間的同步礼预，不再需要另外的對(duì)象變量存儲(chǔ)狀態(tài)，并且也不需要考慮同步鎖虏劲，wait與notify要保證必須有鎖才能執(zhí)行托酸，而且執(zhí)行notify操作釋放鎖后還要將當(dāng)前線程扔進(jìn)該對(duì)象鎖的等待隊(duì)列，LockSupport則完全不用考慮對(duì)象柒巫、鎖励堡、等待隊(duì)列等問(wèn)題。

參考文檔

1. 若文章中的圖片查看不清晰堡掏，請(qǐng)點(diǎn)擊這兒查看大圖