深刻理解Java併發框架AQS系列（三）：獨佔鎖（Exclusive Lock）

時間  2021-03-25
標籤 html java api 數據結構 多線程 架構 併發 框架 ide 高併發 欄目 Java 简体版
原文   原文鏈接

深刻理解Java併發框架AQS系列（一）：線程
深刻理解Java併發框架AQS系列（二）：AQS框架簡介及鎖概念
深刻理解Java併發框架AQS系列（三）：獨佔鎖（Exclusive Lock）html

1、前言

優秀的源碼就在那裏java

通過了前面兩章的鋪墊，終於要切入正題了，本章也是整個AQS的核心之一api

從本章開始，咱們要精讀AQS源碼，在欣賞它的同時也要學會質疑它。固然本文不會帶着你們逐行過源碼（會有「只在此山中，雲深不知處」的弊端），而是從功能入手，對其架構進行逐層剖析，在覈心位置重點解讀，並提出質疑；雖然AQS源碼讀起來比較「跳」，但我仍是建議你們花時間及精力去好好讀它數據結構

本章咱們採用經典併發類ReentrantLock來闡述獨佔鎖多線程

2、總體回顧

獨佔鎖，顧名思義，即在同一時刻，僅容許一個線程執行同步塊代碼。比如一夥兒人想要過河，但只有一根獨木橋，且只能承受一人的重量架構

相信咱們平時寫獨佔鎖的程序大抵是這樣的：併發

ReentrantLock lock = new ReentrantLock();
try {
  lock.lock();
  doBusiness();
} finally {
  lock.unlock();
}

上述代碼分爲三部分：框架

  • 加鎖 lock.lock()
  • 執行同步代碼 doBusiness()
  • 解鎖 lock.unlock()

加鎖部分，必定是衆矢之的，兵家爭搶的要地，對於高併發的程序來講，同一時刻，大量的線程爭相涌入，而lock()則保證只能有一個線程進入doBusiness()邏輯，且在其執行完畢unlock()方法以前，不能有其餘線程進入。因此相對而言，unlock()方法相對輕鬆，不用處理多線程的場景ide

2.一、waitStatus

本章中，咱們引入節點中一個關鍵的字段waitStatus（後文簡寫爲ws），在獨佔鎖模式中，可能會使用到的等待狀態以下：高併發

  • 一、0
    • 初始狀態，當一個節點新建時，其默認ws爲0
  • 二、SIGNAL (-1)
    • 若是某個節點的狀態爲SIGNAL，即代表其後續節點處於（或即將處於）阻塞狀態。因此當前節點在執行完同步代碼或被取消後，必定要記得喚醒其後續節點
  • 三、CANCELLED (1)
    • 顧名思義，即取消操做的含義。當一個節點等待超時、或者被打斷、或者執行tryAcquire發生異常，都會致使當前節點取消。而當節點一旦取消，便永遠不會再變爲0或者SIGNAL狀態了

3、加鎖（核心）

咱們先上一張ReentrantLock加鎖功能（非公平）的總體流程圖，在併發或關鍵部分有註釋

第一眼看上去，確實有點複雜，不過不用怕，咱們逐一分析解讀後，它其實就是隻紙老虎

大致上能夠分爲三大部分

  • a、加入阻塞隊列
  • b、阻塞隊列調度
  • c、異常處理

按照正常的理解，可能只會有a、b兩部分就夠了，爲何會有c呢？何時會發生異常？

3.一、加入阻塞隊列

當一個線程嘗試加鎖失敗後，便會放入阻塞隊列的隊尾；這節咱們來討論一下這個動做的細節

在加入阻塞隊列以前，首先會查看頭節點是否爲null，若是是null的話，須要新建ws爲0的頭結點，（爲何在AQS初始化的時候，不直接新建頭結點呢？其實因而可知做者細節處理的嚴謹，由於若是當咱們的獨佔鎖併發度不大，在嘗試加鎖的過程當中，總能獲取到鎖，這時便不會向阻塞隊列添加內容，假如初始化便新建頭結點，會致使其白白佔用內存空間而得不到有效利用）而後將當前節點添加至阻塞隊列的尾部，固然頭結點初始化、向尾部節點追加新節點都是經過CAS操做的。而阻塞隊列呢，正如咱們前文說起的是一個FIFO的隊列，且帶有next、prev兩個引用來標記前、後節點；咱們在阻塞隊列中加入第一個節點後，阻塞隊列的樣子：

3.二、阻塞隊列調度

這一節屬於獨佔鎖很核心的部分，裏面涉及ws更改、線程掛起與喚醒、更換頭結點等

咱們接着3.1繼續，在節點進入調度後，首先檢查下當前節點的前節點是否爲head節點，若是是的話，那麼有一次嘗試加鎖的機會，加鎖成功或失敗將致使2個分支

咱們首先看加鎖加鎖成功的狀況，一旦加鎖成功，當前節點便從阻塞隊列中「消失」（實際上是當前節點變爲了頭結點，而原頭結點內存不可達，等待垃圾回收），當全部節點都加鎖成功，阻塞隊列便爲空了，但並不表明阻塞隊列的長度爲0，由於有頭結點的存在，因此空阻塞隊列的長度是1

而加鎖失敗或者當前節點的前節點不是head節點呢？是立刻將線程掛起嗎？答案是不肯定的，要看前節點的ws狀態而定。而此步驟還有個隱藏任務：將當前節點以前的全部已取消節點從阻塞隊列中剔除。

從上圖中咱們看到，一個節點若是想正常進入掛起狀態，那麼必定要將前節點的ws改成SIGNAL (-1)狀態，但若是前節點已經變爲CANCELLED (1)狀態後，就要遞歸向前尋找第一個非CANCELLED的節點。

針對「線程掛起並等待其餘線程喚醒」，咱們提出2個問題

問題1

  • 若是是普通節點，直接掛在隊尾，且將其線程掛起，這個沒啥問題；但若是是頭節點被喚醒，嘗試加鎖卻失敗了，又被再次掛起，會不會致使頭結點永遠處於掛起狀態？
  • 答：不會，由於頭結點之因此搶鎖失敗，必定是由於另一個A線程搶鎖成功。雖然頭節點暫時處於掛起狀態，但當A線程執行完加鎖代碼後，還會再次喚醒頭結點

問題2

  • 假定當前節點斷定須要被掛起，在執行掛起操做前，擁有鎖的線程執行完畢，並喚醒了當前線程，而當前線程又立刻要進行掛起操做，豈不是會致使沒法成功將當前節點喚醒，從而永遠hang死？
  • 答：能考慮到這個問題，說明你已經帶着分身去思考問題了，不錯。不過此處是不會存在這個問題的，由於線程掛起、喚醒使用的api爲park/unpark，即使是unpark發生在park以前，在執行park操做時，也會成功喚醒。這個特質區別於wait/notify

而針對阻塞隊列的調度，還有一些沒有解釋的問題：

  • a、爲何阻塞隊列內有這麼多CANCELLED狀態的節點？
  • b、當前節點在掛起前，前節點爲SIGNAL狀態，但通過一段時間運行，前節點變爲了CANCELLED狀態，豈不是致使當前節點永遠沒法被喚醒？

要回答這兩個問題，就要引出異常處理了

3.三、異常處理

咱們首先討論若是AQS不作異常處理能夠嗎？ 不能夠，例如第一個節點被喚醒後，在加鎖階段發生了異常，若是沒有異常處理，這個異常節點將永遠處於阻塞隊列，成爲「殭屍節點」，且後續節點也不會被喚起

官方標明可能會出現異常的部分，諸如「等待超時」、「打斷」等，那若是咱們調用acquire()方法，而非acquireInterruptibly()、tryAcquireNanos(time)是否是就不會出現異常？不是的，由於還有AQS下放給咱們本身實現的tryRelease()等方法。咱們實現一個本身的AQS，並模擬tryRelease()報錯，看AQS可否正常應對

public class FindBugAQS {

  public volatile static int FLAG = 0;

  private static ThreadLocal<Integer> FLAG_STORE = new ThreadLocal<>();

  private static ThreadLocal<Integer> TIMES = ThreadLocal.withInitial(() -> 0);

  private Sync sync = new Sync();

  private static class Sync extends AbstractQueuedSynchronizer {

    private Sync() {
      setState(1);
    }

    public void lock() {
      FLAG_STORE.set(++FLAG);
      int state = getState();
      if (state == 1 && compareAndSetState(state, 0)) {
        return;
      }
      acquire(1);
    }

    @Override
    protected boolean tryAcquire(int acquires) {
      if (FLAG_STORE.get() == 2) {
        Integer time = TIMES.get();
        if (time == 0) {
          TIMES.set(1);
        } else {
          // 模擬發生異常，第二個節點在第二次訪問tryAcquire方法時，將會扔出運行期異常
          System.out.println("發生異常");
          throw new RuntimeException("lkn aqs bug");
        }
      }
      int state = getState();
      if (state == 1 && compareAndSetState(state, 0)) {
        return true;
      }
      return false;
    }

    @Override
    protected final boolean tryRelease(int releases) {
      setState(1);
      return true;
    }

    public void unlock() {
      release(1);
    }
  }

  public void lock() {
    sync.lock();
  }

  public void unlock() {
    sync.unlock();
  }

}

// 測試用例以下：

public class BugTest {
  private static volatile int number = 0;

  @Test
  public void test2() throws InterruptedException {
    List<Thread> list = Lists.newArrayList();
    FindBugAQS aqs = new FindBugAQS();
    Thread thread1 = new Thread(() -> {
      aqs.lock();
      PubTools.sleep(5000);
      number++;
      aqs.unlock();
    });
    thread1.start();
    list.add(thread1);

    PubTools.sleep(500);

    for (int i = 0; i < 4; i++) {
      Thread thread2 = new Thread(() -> {
        aqs.lock();
        PubTools.sleep(500);
        number++;
        aqs.unlock();
      });
      thread2.start();
      list.add(thread2);
    }

    for (Thread thread : list) {
      thread.join();
    }
    System.out.println("number is " + number);
  }
}

運行結果：

發生異常
Exception in thread "Thread-1" java.lang.RuntimeException: lkn aqs bug
	at org.xijiu.share.aqs.bug.FindBugAQS$Sync.tryAcquire(FindBugAQS.java:42)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:863)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1199)
	at org.xijiu.share.aqs.bug.FindBugAQS$Sync.lock(FindBugAQS.java:31)
	at org.xijiu.share.aqs.bug.FindBugAQS.lock(FindBugAQS.java:64)
	at org.xijiu.share.aqs.bug.BugTest.lambda$test2$2(BugTest.java:61)
	at java.lang.Thread.run(Thread.java:748)
number is 4

咱們自定義了AQS實現類FindBugAQS.java，模擬第二個節點在第二次訪問tryAcquire會扔出異常；而後啓動5個線程，對number進行累加。可見，最後的結果符合預期，AQS處理的很完美。那程序發生異常後，阻塞隊列究竟如何應對？

舉例說明吧，假定如今除去頭結點外，阻塞隊列中還有3個節點，當第1個節點被喚醒執行時，發生了異常，那麼第1個節點會將ws置爲CANCELLED，且將向後的鏈條打斷（指向本身），但向前鏈條保持不變，並喚醒下一個節點

由上圖可見，當某個節點響應中斷/發生異常後，其會主動打斷向後鏈條，但依舊保留向前的鏈條，這樣作的目的是爲了後續節點在尋找前節點時，能夠找到標記爲CANCELLED狀態的節點，而不是找到null。至此便解答了3.2提出的兩個問題

a、爲何阻塞隊列內有這麼多CANCELLED狀態的節點？

  • 當被調度執行的節點發生了異常，狀態便會更改成CANCELLED狀態，但仍存在於阻塞隊列中，直到正常執行的節點將其剔除

b、當前節點在掛起前，前節點爲SIGNAL狀態，但通過一段時間運行，前節點變爲了CANCELLED狀態，豈不是致使當前節點永遠沒法被喚醒？

  • 不會，節點發生異常後，會主動喚起後續節點，然後續節點負責將前節點從阻塞隊列中刪除

4、解鎖

原本想針對「解鎖邏輯」畫一張流程圖，但猛然發現解鎖部分僅僅10行左右的代碼，那就索性把源碼貼上，逐一論述下

  • AQS解鎖源碼
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}
  • ReentrantLock解鎖源碼
protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

咱們發現當tryRelease()方法返回true時，AQS便會負責喚醒後續節點，由於ReentrantLock支持了可重入的特性，因此當前線程的每次加鎖都會對state累加，而每次tryRelease()方法則會對state累減，直到state變爲初始狀態0時，tryRelease()方法纔會返回true，即喚醒下一個節點

解鎖邏輯相對簡潔，且不存在併發，本文再也不贅述

5、後記

再次強調本文是經過ReentrantLock的視角來分析獨佔鎖，且主要分析的是ReentrantLock.lock()/unlock()方法，目的是讓你們對AQS總體的數據結構有個全面認識，方便後續在實現本身的併發框架時，明白api背後發生的事情，作到遊刃有餘

而像ReentrantLock的lockInterruptibly()、tryLock(TimeUnit)或者其餘獨佔鎖的實現類，讀者可自行閱讀源碼，原理相似，核心代碼也是同樣的

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程