ReentrantLock源碼一覽

時間 2019-11-09

原文原文鏈接

大綱

前言

ReentrantLock，在面試的時候也是高頻問題。java

他是一個可重入鎖(一樣的還有 Synchronized) ，可重入的意思是，當一個資源被一個線程拿到並加了鎖以後，這個線程還能夠再次進入這個資源，而後再次加鎖。鎖的count++。當count==0，以後，才表示這個鎖被釋放，其餘線程能夠搶奪node

那麼他是怎麼實現的呢？git

咱們進入到ReentrantLock的源碼中進行分析github

ps:

全部文章同步更新與Github--Java-Notes,想了解JVM(基本更完），HashMap源碼分析，spring相關，併發，劍指offer題解（Java版），能夠點個star。能夠看個人github主頁，天天都在更新喲(自從上班，天天都是晚上寫到12點多，早上6點多起來碼字,天天的動力就是看這star數往上漲)。面試

繼承體系

若是不知道這個怎麼看或者不知道這個怎麼調出來的，能夠看看個人這篇文章，看源碼用到的小工具spring

從圖中咱們能夠看到，ReentrantLock實現了兩個接口，Lock和Serializable。其中Lock接口裏面定義了鎖要使用的方法編程

而後ReentrantLock裏面還有一個重要的抽象類 Sync，它定義了不少方法。安全

咱們經過最開始的繼承圖片能夠看出來，這個Sync類繼承自AQS(AbstactQueuedSynchronizer)，請記住這個AQS，由於他是JUC ，Java併發工具中的核心。之後會常常出現，面試也會問這個玩意兒併發

再看AQS app

他這裏定義了不少不少東西，由於實在太多，並且很重要，因此我又把它單獨抽出來，封裝成了另外一篇文章談談AQS

從哪裏入手

我是看的《Java併發編程的藝術》，就以他的順序來，從lock入手

獲取鎖

acquire

// 他又去調用 sync 實例的lock方法，
//咱們以前說過，不少的處理邏輯都是在Sync這個類中完成的
public void lock() {
    sync.lock();
}


  /** * Performs {@link Lock#lock}. The main reason for subclassing * is to allow fast path for nonfair version */
// 可是這個類中是一個抽象方法，緣由也寫的很清楚了
// 主要緣由是爲了非公平鎖的子類能更快的找到他的非公平處理方法
abstract void lock();
複製代碼

/** * Sync object for fair locks */
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock() {
      // 調用AQS中的方法
        acquire(1);
    }
  
  

  // acquire 是 AQS的方法
  public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
      // 若是既沒有獲取到鎖，也沒有將其加入隊列，則此線程中斷
      selfInterrupt();
  }
  
  /**再往下看tryAcquire方法，這個方法AQS裏沒有實現，直接拋出了異常，這麼作是避免子類實現全部接口，咱們看java.util.concurrent.locks.ReentrantLock.FairSync這個AQS子類的實現 */
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

    /** * Fair version of tryAcquire. Don't grant access unless * recursive call or no waiters or is first. */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
      	// 獲取到線程的狀態，0表示釋放了鎖，state>0表示獲取了鎖
        int c = getState();
        if (c == 0) {
          // 若是沒有其餘線程佔用，而且經過CAS操做把線程狀態設置爲1了，
          // 那麼就將線程指向當前線程
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
      // 若是 此線程是重入的，即佔有這個資源的仍是原來的那把鎖，則將計數器+1
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}
複製代碼

acquireQueued

剛剛咱們看了前面的一部分，在看邏輯與運算的的後半截

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
複製代碼

他調用了acquireQueued的方法，而且傳入了兩個參數，一個是 addWaiter方法，一個是以前傳過來的參數 1.

addWaiter方法傳入了一個 獨佔式(EXCLUSIVE)的Node(還有共享式的 SHARED)

咱們進入這個方法

/** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared * @return the new node */

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
  	// 隊列的末尾不爲空，即有線程拿到了鎖，就CAS入隊
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}
複製代碼

若是隊列中沒有線程，那麼就調用下面的enq方法

/** * Inserts node into queue, initializing if necessary. See picture above. * @param node the node to insert * @return node's predecessor */
// 一直循CAS設置頭和尾，直到兩個都成功。
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
複製代碼

而後咱們再看acquireQueued方法

/** * Acquires in exclusive uninterruptible mode for thread already in * queue. Used by condition wait methods as well as acquire. * * @param node the node * @param arg the acquire argument * @return {@code true} if interrupted while waiting */
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
          /* * 若是前置節點是head，說明當前節點是隊列第一個等待的節點，這時去 * 嘗試獲取鎖，若是成功了則 * 獲取鎖成功。這裏有的同窗可能沒看懂，不是剛嘗試失敗並插入隊列 * 嗎，咋又嘗試獲取鎖？ 其實這 * 裏是個循環，其餘剛被喚醒的線程也會執行到這個代碼 */
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
           // 判斷 獲取失敗是否須要線程阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
複製代碼

咱們再進入 shouldParkAfterFailedAcquire 這個方法，

/** * Checks and updates status for a node that failed to acquire. * Returns true if thread should block. This is the main signal * control in all acquire loops. Requires that pred == node.prev. * * @param pred node's predecessor holding status * @param node the node * @return {@code true} if thread should block */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 找到前驅節點的狀態
    int ws = pred.waitStatus;
    // 若是前驅節點的狀態 爲 SINGAl，那麼本次的阻塞能夠安全進行
  	// 由於前驅節點承諾執行完後會喚醒當前線程
    if (ws == Node.SIGNAL)
        /* * This node has already set status asking a release * to signal it, so it can safely park. */
        return true;
  	// 若是前驅節點大於0，那麼說明他已經取消了，要往前遍歷鏈表，找到不是已取消狀態的節點，並將其後繼節點設置爲傳進來的當前節點
    if (ws > 0) {
        /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don't park yet. Caller will need to * retry to make sure it cannot acquire before parking. */
      	// 若是都不是，那麼就說明 狀態不是 0 就是 PROPAGATE，
      	// 這樣的話調用者再嘗試一下可否獲取鎖
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
複製代碼

public static void park(Object blocker) {
    Thread t = Thread.currentThread();
  	// 設置當前線程的 block 
    setBlocker(t, blocker);
  	// 使用UNSAFE 類，也就是native方法到JVM級別去阻塞當前線程
    UNSAFE.park(false, 0L);
  	// 將 blocker 設置成 空
    setBlocker(t, null);
}
複製代碼

State的各個狀態

Lock步驟總結

調用tryAcquire方法嘗試獲取鎖，獲取成功的話修改state並直接返回true，獲取失敗的話把當前線程加到等待隊列中
加到等待隊列以後先檢查前置節點狀態是不是signal，若是是的話直接阻塞當前線程等待喚醒，若是不是的話判斷是不是cancel狀態，是cancel狀態就往前遍歷並把cancel狀態的節點從隊列中刪除。若是狀態是0或者propagate的話將其修改爲signal
阻塞被喚醒以後若是是隊首而且嘗試獲取鎖成功就返回true，不然就繼續執行前一步的代碼進入阻塞

釋放鎖

// 先調用 ReentrantLock 的 unlock方法
// 這個方法會調用內部類的release方法，可是這個方法是Sync從父類AQS繼承過來的
// 因此他是調用的 AQS裏面的release方法
public void unlock() {
    sync.release(1);
}
複製代碼

咱們再次進入AQS ，找到release方法

release

/** * Releases in exclusive mode. Implemented by unblocking one or * more threads if {@link #tryRelease} returns true. * This method can be used to implement method {@link Lock#unlock}. * * @param arg the release argument. This value is conveyed to * {@link #tryRelease} but is otherwise uninterpreted and * can represent anything you like. * @return the value returned from {@link #tryRelease} */

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
      	// 若是鏈表有頭，而且不爲 0 ，就喚醒後面的節點
        if (h != null && h.waitStatus != 0)
          // 釋放這個節點
            unparkSuccessor(h);
        return true;
    }
    return false;
}
複製代碼

protected final boolean tryRelease(int releases) {
  	// 獲得鎖的計數器
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
  	// 若是state爲0，說明已經解鎖了
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}
複製代碼

unparkSuccessor

/** * Wakes up node's successor, if one exists. * * @param node the node */

// 喚醒後繼節點
private void unparkSuccessor(Node node) {
    /* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */
    int ws = node.waitStatus;
    if (ws < 0)
      	// 使用CAS設置等待狀態
        compareAndSetWaitStatus(node, ws, 0);

    /* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled successor. */ 
    Node s = node.next;
    // 若是當前節點的後繼節點爲空 或者是取消狀態，從後面遍歷，
  	// 找到不是取消狀態的，並將其設置爲後繼節點
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
  	// 若是後繼節點不爲空，喚醒後繼節點
    if (s != null)
        LockSupport.unpark(s.thread);
}
複製代碼