Java併發編程-ReentrantLock_ZenDei技術網路在線

上文已經總結了AQS的前世今生，有了這個基礎我們就可以來進一步學習併發工具類。首先我們要學習的就是ReentrantLock，本文將從ReentrantLock的產生背景、源碼原理解析和應用來學習ReentrantLock這個併發工具類。 1、產生背景前面我們已經學習過了synchronized ...

　　上文已經總結了AQS的前世今生，有了這個基礎我們就可以來進一步學習併發工具類。首先我們要學習的就是ReentrantLock，本文將從ReentrantLock的產生背景、源碼原理解析和應用來學習ReentrantLock這個併發工具類。

1、 產生背景

　　前面我們已經學習過了synchronized，這個關鍵字可以確保對象在併發訪問中的原子性、可見性和有序性，這個關鍵字的底層交由了JVM通過C++來實現，既然是JVM實現，就依賴於JVM，程式員就無法在Java層面進行擴展和優化，肯定就靈活性不高，比如程式員在使用時就無法中斷一個正在等待獲取鎖的線程，或者無法在請求一個鎖時無限的等待下去。基於這樣一個背景，Doug Lea構建了一個在記憶體語義上和synchronized一樣效果的Java類，同時還擴展了其他一些高級特性，比如定時的鎖等待、可中斷的鎖等待和公平性等，這個類就是ReentrantLock。

2、 源碼原理解析

2.1 可重入性原理

　　在synchronized一文中，我們認為synchronized是一種重量級鎖，它的實現對應的是C++的ObjectMonitor，代碼如下：

ObjectMonitor() {
    _header       = NULL;
    _count        = 0; //記錄線程獲取鎖的次數
    _waiters      = 0;
    _recursions   = 0; //鎖的重入次數
    _object       = NULL;
    _owner        = NULL;//指向持有ObjectMonitor對象的線程
    _WaitSet      = NULL; //等待條件隊列 類似AQS的ConditionObject
    _WaitSetLock  = 0 ;
    _Responsible  = NULL ;
    _succ         = NULL ;
    _cxq          = NULL ;
    FreeNext      = NULL ;
    _EntryList    = NULL ; //同步隊列 類似AQS的CLH隊列
    _SpinFreq     = 0 ;
    _SpinClock    = 0 ;
    OwnerIsThread = 0 ;
    _previous_owner_tid = 0;
  }

　　從代碼中可以看到synchronized實現的鎖的重入依賴於JVM，JVM為每個對象的鎖關聯一個計數器_count和一個所有者線程_owner，當計數器為0的時候就認為鎖沒有被任何線程持有，當線程請求一個未被持有的鎖時，JVM就記下鎖的持有者，並將計數器的值設置為1，如果是同一個線程再次獲取這個鎖，計數器的值遞增，而當線程退出時，計數器的值遞減，直到計數器為0時，鎖被釋放。

　　ReentrantLock實現了在記憶體語義上的synchronized，固然也是支持可重入的，那麼ReentrantLock是如何支持的呢，讓我們以非公平鎖的實現看下ReentrantLock的可重入，代碼如下：

final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();//當前線程
            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");
                setState(nextc);
                return true;
            }
            return false;
        }

        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;//只有線程全部釋放才返回true
                setExclusiveOwnerThread(null); //同步隊列的線程都可以去獲取同步狀態了
            }
            setState(c); 
            return free;
        }

　　看到這也就明白了上文說的ReentrantLock類使用AQS同步狀態來保存鎖重覆持有的次數。當鎖被一個線程獲取時，ReentrantLock也會記錄下當前獲得鎖的線程標識，以便檢查是否是重覆獲取，以及當錯誤的線程試圖進行解鎖操作時檢測是否存在非法狀態異常。

2.2 獲取和釋放鎖

　　如下是獲取和釋放鎖的方法：

public void lock() {
   sync.lock();//獲取鎖
}
public void unlock() {
   sync.release(1); //釋放鎖
}

　　獲取鎖的時候依賴的是內部類Sync的lock()方法，該方法又有2個實現類方法，分別是非公平鎖NonfairSync和公平鎖FairSync，具體咱們下一小節分析。再來看下釋放鎖，釋放鎖的時候實際調用的是AQS的release方法，代碼如下：

public final boolean release(int arg) {
        if (tryRelease(arg)) {//調用子類的tryRelease 實際就是Sync的tryRelease
            Node h = head;//取同步隊列的頭節點
            if (h != null && h.waitStatus != 0)//同步隊列頭節點不為空且不是初始狀態
                unparkSuccessor(h);//釋放頭節點 喚醒後續節點
            return true;
        }
        return false;
}

　　Sync的tryRelease就是上一小節的重入釋放方法，如果是同一個線程，那麼鎖的重入次數就依次遞減，直到重入次數為0，此方法才會返回true,此時斷開頭節點喚醒後續節點去獲取AQS的同步狀態。

2.3 公平鎖和非公平鎖

　　公平鎖還是非公平鎖取決於ReentrantLock的構造方法，預設無參構造方法是NonfairSync，含參構造方法，入參true為FairSync，入參false為NonfairSync。

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

　　再分別來看看非公平鎖和公平鎖的實現。

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
        final void lock() {
            if (compareAndSetState(0, 1))//通過CAS來獲取同步狀態 也就是鎖
                setExclusiveOwnerThread(Thread.currentThread());//獲取成功線程占有鎖
            else
                acquire(1);//獲取失敗 進入AQS同步隊列排隊等待 執行AQS的acquire方法 
        }

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }

　　在AQS的acquire方法中先調用子類tryAcquire，也就是nonfairTryAcquire，見2.1小節。可以看出非公平鎖中，搶到AQS的同步狀態的未必是同步隊列的首節點，只要線程通過CAS搶到了同步狀態或者在acquire中搶到同步狀態，就優先占有鎖，而相對同步隊列這個嚴格的FIFO隊列來說，所以會被認為是非公平鎖。

static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;

        final void lock() {
            acquire(1);//嚴格按照AQS的同步隊列要求去獲取同步狀態
        }

        /**
         * 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();//獲取當前線程
            int c = getState();
            if (c == 0) {//鎖未被搶占
                if (!hasQueuedPredecessors() &&//沒有前驅節點
                    compareAndSetState(0, acquires)) {//CAS獲取同步狀態
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {//鎖已被搶占且線程重入
                int nextc = c + acquires;//同步狀態為重入次數
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

　　公平鎖的實現直接調用AQS的acquire方法，acquire中調用tryAcquire。和非公平鎖相比，這裡不會執行一次CAS，接下來在tryAcquire去搶占鎖的時候，也會先調用hasQueuedPredecessors看看前面是否有節點已經在等待獲取鎖了，如果存在則同步隊列的前驅節點優先。

public final boolean hasQueuedPredecessors() {
        // The correctness of this depends on head being initialized
        // before tail and on head.next being accurate if the current
        // thread is first in queue.
        Node t = tail; // Read fields in reverse initialization order 尾節點
        Node h = head;//頭節點
        Node s;
        return h != t &&//頭尾節點不是一個 即隊列存在排隊線程
            ((s = h.next) == null || s.thread != Thread.currentThread());//頭節點的後續節點為空或者不是當前線程
    }

　　雖然公平鎖看起來在公平性上比非公平鎖好，但是公平鎖為此付出了大量線程切換的代價，而非公平鎖在鎖的獲取上不能保證公平，就有可能出現鎖饑餓，即有的線程多次獲取鎖而有的線程獲取不到鎖，沒有大量的線程切換保證了非公平鎖的吞吐量。

3、應用

3.1普通的線程鎖

　　標準形式如下：

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

　　這種用法和synchronized效果是一樣的，但是必須顯示的聲明lock和unlock。

3.2 帶限制的鎖

public boolean tryLock()// 嘗試獲取鎖,立即返回獲取結果 輪詢鎖
public boolean tryLock(long timeout, TimeUnit unit)//嘗試獲取鎖,最多等待 timeout 時長 超時鎖
public void lockInterruptibly()//可中斷鎖,調用線程 interrupt 方法,則鎖方法拋出 InterruptedException  中斷鎖

　　具體可查看github鏈接裡面的ReentrantLockTest。

3.3 等待/通知模型

　　內置隊列存在一些缺陷，每個內置鎖只能關聯一個條件隊列(_WaitSet)，這導致多個線程可能會在同一個條件隊列上等待不同的條件謂詞，如果每次使用notify喚醒條件隊列，可能會喚醒錯誤的線程導致喚醒失敗，但是如果使用notifyAll的話，能喚醒到正確的線程，因為所有的線程都會被喚醒，這也帶來一個問題，就是不應該被喚醒的在被喚醒後發現不是自己等待的條件謂詞轉而又被掛起。這樣的操作會帶來系統的資源浪費，降低系統性能。這個時候推薦使用顯式的Lock和Condition來替代內置鎖和條件隊列，從而控制多個條件謂詞的情況，達到精確的控制線程的喚醒和掛起。具體後面再來分析下JVM的內置鎖、條件隊列模型和顯式的Lock、Condition模型，實際上在AQS裡面也提到了Lock、Condition模型。

3.4 和synchronized比較

　　兩者的區別大致如下：

synchronized	ReentrantLock
使用Object本身的wait、notify、notifyAll調度機制	與Condition結合進行線程的調度
顯式的使用在同步方法或者同步代碼塊	顯式的聲明指定起始和結束位置
托管給JVM執行，不會因為異常、或者未釋放而發生死鎖	手動釋放鎖

　　Jdk1.6之前，ReentrantLock性能優於synchronized，不過1.6之後，synchronized做了大量的性能調優，而且synchronized相對程式員來說，簡潔熟悉，如果不是synchronized無法實現的功能，如輪詢鎖、超時鎖和中斷鎖等，推薦首先使用synchronized，而針對鎖的高級功能，再使用ReentrantLock。

參考資料:

https://github.com/lingjiango/ConcurrentProgramPractice

https://www.ibm.com/developerworks/cn/java/j-jtp10264/index.html

https://www.ibm.com/developerworks/java/library/j-jtp10264/