ReentrantLock內部原理

時間  2019-12-05
標籤 reentrantlock 內部 原理 欄目 Java 简体版
原文   原文鏈接

1 synchronized和lock

1.1 synchronized的侷限性

    synchronized是java內置的關鍵字，它提供了一種獨佔的加鎖方式。synchronized的獲取和釋放鎖由JVM實現，用戶不須要顯示的釋放鎖，很是方便。然而synchronized也有必定的侷限性，例如：java

  1. 當線程嘗試獲取鎖的時候，若是獲取不到鎖會一直阻塞。
  2. 若是獲取鎖的線程進入休眠或者阻塞，除非當前線程異常，不然其餘線程嘗試獲取鎖必須一直等待。

    JDK1.5以後發佈，加入了Doug Lea實現的concurrent包。包內提供了Lock類，用來提供更多擴展的加鎖功能。Lock彌補了synchronized的侷限，提供了更加細粒度的加鎖功能。node

1.2 Lock簡介

Lock api以下編程

 

void lock();
void lockInterruptibly() throws InterruptedException;
boolean tryLock();
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
void unlock();
Condition newCondition();

 

其中最經常使用的就是lock和unlock操做了。由於使用lock時，須要手動的釋放鎖，因此須要使用try..catch來包住業務代碼，而且在finally中釋放鎖。典型使用以下api

 

private Lock lock = new ReentrantLock();
 
public void test(){
    lock.lock();
    try{
        doSomeThing();
    }catch (Exception e){
        // ignored
    }finally {
        lock.unlock();
    }
}

 

2 AQS

    AbstractQueuedSynchronizer簡稱AQS，是一個用於構建鎖和同步容器的框架。事實上concurrent包內許多類都是基於AQS構建，例如ReentrantLock，Semaphore，CountDownLatch，ReentrantReadWriteLock，FutureTask等。AQS解決了在實現同步容器時設計的大量細節問題。併發

    AQS使用一個FIFO的隊列表示排隊等待鎖的線程，隊列頭節點稱做「哨兵節點」或者「啞節點」，它不與任何線程關聯。其餘的節點與等待線程關聯，每一個節點維護一個等待狀態waitStatus。如圖框架

     AQS中還有一個表示狀態的字段state，例如ReentrantLocky用它表示線程重入鎖的次數，Semaphore用它表示剩餘的許可數量，FutureTask用它表示任務的狀態。對state變量值的更新都採用CAS操做保證更新操做的原子性。ui

    AbstractQueuedSynchronizer繼承了AbstractOwnableSynchronizer，這個類只有一個變量：exclusiveOwnerThread，表示當前佔用該鎖的線程，而且提供了相應的get，set方法。this

    理解AQS能夠幫助咱們更好的理解JCU包中的同步容器。spa

3 lock()與unlock()實現原理

3.1 基礎知識

    ReentrantLock是Lock的默認實現之一。那麼lock()和unlock()是怎麼實現的呢？首先咱們要弄清楚幾個概念線程

  • 可重入鎖。可重入鎖是指同一個線程能夠屢次獲取同一把鎖。ReentrantLock和synchronized都是可重入鎖。
  • 可中斷鎖。可中斷鎖是指線程嘗試獲取鎖的過程當中，是否能夠響應中斷。synchronized是不可中斷鎖，而ReentrantLock則提供了中斷功能。
  • 公平鎖與非公平鎖。公平鎖是指多個線程同時嘗試獲取同一把鎖時，獲取鎖的順序按照線程達到的順序，而非公平鎖則容許線程「插隊」。synchronized是非公平鎖，而ReentrantLock的默認實現是非公平鎖，可是也能夠設置爲公平鎖。
  • CAS操做(CompareAndSwap)。CAS操做簡單的說就是比較並交換。CAS 操做包含三個操做數 —— 內存位置（V）、預期原值（A）和新值(B)。若是內存位置的值與預期原值相匹配，那麼處理器會自動將該位置值更新爲新值。不然，處理器不作任何操做。不管哪一種狀況，它都會在 CAS 指令以前返回該位置的值。CAS 有效地說明了「我認爲位置 V 應該包含值 A；若是包含該值，則將 B 放到這個位置；不然，不要更改該位置，只告訴我這個位置如今的值便可。」 Java併發包(java.util.concurrent)中大量使用了CAS操做,涉及到併發的地方都調用了sun.misc.Unsafe類方法進行CAS操做。

3.2 內部結構

    ReentrantLock提供了兩個構造器，分別是

 

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

 

    默認構造器初始化爲NonfairSync對象，即非公平鎖，而帶參數的構造器能夠指定使用公平鎖和非公平鎖。由lock()和unlock的源碼能夠看到，它們只是分別調用了sync對象的lock()和release(1)方法。

    Sync是ReentrantLock的內部類，它的結構以下

 能夠看到Sync擴展了AbstractQueuedSynchronizer。

3.3 NonfairSync

    咱們從源代碼出發，分析非公平鎖獲取鎖和釋放鎖的過程。 

3.3.1 lock() 

    lock()源碼以下

 

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

 

    首先用一個CAS操做，判斷state是不是0（表示當前鎖未被佔用），若是是0則把它置爲1，而且設置當前線程爲該鎖的獨佔線程，表示獲取鎖成功。當多個線程同時嘗試佔用同一個鎖時，CAS操做只能保證一個線程操做成功，剩下的只能乖乖的去排隊啦。

    「非公平」即體如今這裏，若是佔用鎖的線程剛釋放鎖，state置爲0，而排隊等待鎖的線程還未喚醒時，新來的線程就直接搶佔了該鎖，那麼就「插隊」了。

    若當前有三個線程去競爭鎖，假設線程A的CAS操做成功了，拿到了鎖開開心心的返回了，那麼線程B和C則設置state失敗，走到了else裏面。咱們往下看acquire。

acquire(arg)

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

代碼很是簡潔，可是背後的邏輯卻很是複雜，可見Doug Lea大神的編程功力。

 1. 第一步。嘗試去獲取鎖。若是嘗試獲取鎖成功，方法直接返回。

tryAcquire(arg)

 

final boolean nonfairTryAcquire(int acquires) {
    //獲取當前線程
    final Thread current = Thread.currentThread();
    //獲取state變量值
    int c = getState();
    if (c == 0) { //沒有線程佔用鎖
        if (compareAndSetState(0, acquires)) {
            //佔用鎖成功,設置獨佔線程爲當前線程
            setExclusiveOwnerThread(current);
            return true;
        }
    } else if (current == getExclusiveOwnerThread()) { //當前線程已經佔用該鎖
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        // 更新state值爲新的重入次數
        setState(nextc);
        return true;
    }
    //獲取鎖失敗
    return false;
}

 

    非公平鎖tryAcquire的流程是：檢查state字段，若爲0，表示鎖未被佔用，那麼嘗試佔用，若不爲0，檢查當前鎖是否被本身佔用，若被本身佔用，則更新state字段，表示重入鎖的次數。若是以上兩點都沒有成功，則獲取鎖失敗，返回false。

2. 第二步，入隊。因爲上文中提到線程A已經佔用了鎖，因此B和C執行tryAcquire失敗，而且入等待隊列。若是線程A拿着鎖死死不放，那麼B和C就會被掛起。

先看下入隊的過程。

先看addWaiter(Node.EXCLUSIVE)

 

/**
 * 將新節點和當前線程關聯而且入隊列
 * @param mode 獨佔/共享
 * @return 新節點
 */
private Node addWaiter(Node mode) {
    //初始化節點,設置關聯線程和模式(獨佔 or 共享)
    Node node = new Node(Thread.currentThread(), mode);
    // 獲取尾節點引用
    Node pred = tail;
    // 尾節點不爲空,說明隊列已經初始化過
    if (pred != null) {
        node.prev = pred;
        // 設置新節點爲尾節點
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 尾節點爲空,說明隊列還未初始化,須要初始化head節點併入隊新節點
    enq(node);
    return node;
}

 

B、C線程同時嘗試入隊列，因爲隊列還沒有初始化，tail==null，故至少會有一個線程會走到enq(node)。咱們假設同時走到了enq(node)裏。

 

/**
 * 初始化隊列而且入隊新節點
 */
private Node enq(final Node node) {
    //開始自旋
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            // 若是tail爲空,則新建一個head節點,而且tail指向head
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            // tail不爲空,將新節點入隊
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

 

這裏體現了經典的自旋+CAS組合來實現非阻塞的原子操做。因爲compareAndSetHead的實現使用了unsafe類提供的CAS操做，因此只有一個線程會建立head節點成功。假設線程B成功，以後B、C開始第二輪循環，此時tail已經不爲空，兩個線程都走到else裏面。假設B線程compareAndSetTail成功，那麼B就能夠返回了，C因爲入隊失敗還須要第三輪循環。最終全部線程均可以成功入隊。

     當B、C入等待隊列後，此時AQS隊列以下：

3. 第三步，掛起。B和C相繼執行acquireQueued(final Node node, int arg)。這個方法讓已經入隊的線程嘗試獲取鎖，若失敗則會被掛起。

 

/**
 * 已經入隊的線程嘗試獲取鎖
 */
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); // 獲取成功,將當前節點設置爲head節點
                p.next = null; // 原head節點出隊,在某個時間點被GC回收
                failed = false; //獲取成功
                return interrupted; //返回是否被中斷過
            }
            // 判斷獲取失敗後是否能夠掛起,若能夠則掛起
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                // 線程若被中斷,設置interrupted爲true
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

 

code裏的註釋已經很清晰的說明了acquireQueued的執行流程。假設B和C在競爭鎖的過程當中A一直持有鎖，那麼它們的tryAcquire操做都會失敗，所以會走到第2個if語句中。咱們再看下shouldParkAfterFailedAcquire和parkAndCheckInterrupt都作了哪些事吧。

 

/**
 * 判斷當前線程獲取鎖失敗以後是否須要掛起.
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //前驅節點的狀態
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        // 前驅節點狀態爲signal,返回true
        return true;
    // 前驅節點狀態爲CANCELLED
    if (ws > 0) {
        // 從隊尾向前尋找第一個狀態不爲CANCELLED的節點
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 將前驅節點的狀態設置爲SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
  
/**
 * 掛起當前線程,返回線程中斷狀態並重置
 */
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

 

    線程入隊後可以掛起的前提是，它的前驅節點的狀態爲SIGNAL，它的含義是「Hi，前面的兄弟，若是你獲取鎖而且出隊後，記得把我喚醒！」。因此shouldParkAfterFailedAcquire會先判斷當前節點的前驅是否狀態符合要求，若符合則返回true，而後調用parkAndCheckInterrupt，將本身掛起。若是不符合，再看前驅節點是否>0(CANCELLED)，如果那麼向前遍歷直到找到第一個符合要求的前驅，若不是則將前驅節點的狀態設置爲SIGNAL。

     整個流程中，若是前驅結點的狀態不是SIGNAL，那麼本身就不能安心掛起，須要去找個安心的掛起點，同時能夠再嘗試下看有沒有機會去嘗試競爭鎖。

    最終隊列可能會以下圖所示

  線程B和C都已經入隊，而且都被掛起。當線程A釋放鎖的時候，就會去喚醒線程B去獲取鎖啦。

3.3.2 unlock()

unlock相對於lock就簡單不少。源碼以下

 

public void unlock() {
    sync.release(1);
}
  
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

 

若是理解了加鎖的過程，那麼解鎖看起來就容易多了。流程大體爲先嚐試釋放鎖，若釋放成功，那麼查看頭結點的狀態是否爲SIGNAL，若是是則喚醒頭結點的下個節點關聯的線程，若是釋放失敗那麼返回false表示解鎖失敗。這裏咱們也發現了，每次都只喚起頭結點的下一個節點關聯的線程。

   最後咱們再看下tryRelease的執行過程

 

/**
 * 釋放當前線程佔用的鎖
 * @param releases
 * @return 是否釋放成功
 */
protected final boolean tryRelease(int releases) {
    // 計算釋放後state值
    int c = getState() - releases;
    // 若是不是當前線程佔用鎖,那麼拋出異常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        // 鎖被重入次數爲0,表示釋放成功
        free = true;
        // 清空獨佔線程
        setExclusiveOwnerThread(null);
    }
    // 更新state值
    setState(c);
    return free;
}

 

這裏入參爲1。tryRelease的過程爲：當前釋放鎖的線程若不持有鎖，則拋出異常。若持有鎖，計算釋放後的state值是否爲0，若爲0表示鎖已經被成功釋放，而且則清空獨佔線程，最後更新state值，返回free。 

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