[[386834]]

本文轉載自微信公眾號「三太子敖丙」，作者三太子敖丙 。轉載本文請聯系三太子敖丙公眾號。

使用Java進行多線程開發，使用鎖是一個幾乎不可避免的問題。今天，就讓我們來聊一聊這個基礎，但是又特別特別重要的話題。

首先，讓我們來看一下，到底什么是鎖? 以及，為什么要使用鎖?

如果有2個線程，需要訪問同一個對象User。一個讀線程，一個寫線程。User對象有2個字段，一個是名字，一個是手機號碼。

 

當User對象剛剛創建出來的時候，姓名和手機號碼都是空。然后，寫線程開始填充數據。最后，就出現了以下令人心碎的一幕：

 

可以看到，雖然寫線程先于讀線程工作，但是， 由于寫姓名和寫電話號碼兩個操作不是原子的。這就導致讀線程只讀取了半個數據，在讀線程看來，User對象的電話號碼是不存在。

為了避免類似的問題，我們就需要使用鎖。讓寫線程在修改對象前，先加鎖，然后完成姓名和電話號碼的賦值，再釋放鎖。而讀線程也是一樣，先取得鎖，再讀，然后釋放鎖。這樣就可以避免發生這種情況。

如下圖所示：

 

什么是重入鎖?

好了，現在大家知道我們為什么要使用鎖了。那什么是重入鎖呢。通常情況下，鎖可以用來控制多線程的訪問行為。那對于同一個線程，如果連續兩次對同一把鎖進行lock，會怎么樣了?對于一般的鎖來說，這個線程就會被永遠卡死在那邊，比如：

void handle() { 
    lock(); 
    lock();  //和上一個lock()操作同一個鎖對象，那么這里就永遠等待了 
    unlock(); 
    unlock(); 
} 

這個特性相當不好用，因為在實際的開發過程中，函數之間的調用關系可能錯綜復雜，一個不小心就可能在多個不同的函數中，反復調用lock()，這樣的話，線程就自己和自己卡死了。

所以，對于希望傻瓜式編程的我們來說，重入鎖就是用來解決這個問題的。重入鎖使得同一個線程可以對同一把鎖，在不釋放的前提下，反復加鎖，而不會導致線程卡死。因此，如果我們使用的是重入鎖，那么上述代碼就 可以正常工作。你唯一需要保證的，就是unlock()的次數和lock()一樣多。這樣是不是方便很多呢?

Java中的重入鎖

Java中的鎖都來自與Lock接口，如下圖中紅框內的，就是重入鎖。

 

重入鎖提供的最重要的方法就是lock()

• void lock()：加鎖，如果鎖已經被別人占用了，就無限等待。

這個lock()方法，提供了鎖最基本的功能，拿到鎖就返回，拿不到就等待。因此，大規模得在復雜場景中使用，是有可能因此死鎖的。因此，使用這個方法得非常小心。

如果要預防可能發生的死鎖，可以嘗試使用下面這個方法:

• boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException：嘗試獲取鎖，等待timeout時間。同時，可以響應中斷。

這是一個比單純lock()更具有工程價值的方法，如果大家閱讀過JDK的一些內部代碼，就不難發現，tryLock()在JDK內部被大量的使用。

與lock()相比,tryLock()至少有下面一個好處：

1. 可以不用進行無限等待。直接打破形成死鎖的條件。如果一段時間等不到鎖，可以直接放棄，同時釋放自己已經得到的資源。這樣，就可以在很大程度上，避免死鎖的產生。因為線程之間出現了一種謙讓機制
2. 可以在應用程序這層進行進行自旋，你可以自己決定嘗試幾次，或者是放棄。
3. 等待鎖的過程中可以響應中斷，如果此時，程序正好收到關機信號，中斷就會觸發，進入中斷異常后，線程就可以做一些清理工作，從而防止在終止程序時出現數據寫壞，數據丟失等悲催的情況。

當然了，當鎖使用完后，千萬不要忘記把它釋放了。不然，程序可能就會崩潰啦~

• void unlock() ：釋放鎖

此外， 重入鎖還有一個不帶任何參數的tryLock()。

• public boolean tryLock()

這個不帶任何參數的tryLock()不會進行任何等待，如果能夠獲得鎖，直接返回true，如果獲取失敗，就返回false，特別適合在應用層自己對鎖進行管理，在應用層進行自旋等待。

重入鎖的實現原理

重入鎖內部實現的主要類如下圖：

 

重入鎖的核心功能委托給內部類Sync實現，并且根據是否是公平鎖有FairSync和NonfairSync兩種實現。這是一種典型的策略模式。

實現重入鎖的方法很簡單，就是基于一個狀態變量state。這個變量保存在AbstractQueuedSynchronizer對象中

private volatile int state; 

當這個state==0時，表示鎖是空閑的，大于零表示鎖已經被占用， 它的數值表示當前線程重復占用這個鎖的次數。因此，lock()的最簡單的實現是：

final void lock() { 
// compareAndSetState就是對state進行CAS操作，如果修改成功就占用鎖 
if (compareAndSetState(0, 1)) 
    setExclusiveOwnerThread(Thread.currentThread()); 
else 
//如果修改不成功，說明別的線程已經使用了這個鎖，那么就可能需要等待 
    acquire(1); 

下面是acquire() 的實現：

public final void acquire(int arg) { 
//tryAcquire() 再次嘗試獲取鎖， 
//如果發現鎖就是當前線程占用的，則更新state，表示重復占用的次數， 
//同時宣布獲得所成功,這正是重入的關鍵所在 
if (!tryAcquire(arg) && 
    // 如果獲取失敗，那么就在這里入隊等待 
    acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 
    //如果在等待過程中 被中斷了，那么重新把中斷標志位設置上 
    selfInterrupt(); 

公平的重入鎖

默認情況下，重入鎖是不公平的。什么叫不公平呢。也就是說，如果有1,2,3,4 這四個線程，按順序，依次請求鎖。那等鎖可用的時候，誰會先拿到鎖呢?在非公平情況下，答案是隨機的。如下圖所示，可能線程3先拿到鎖。

 

如果你是一個公平主義者，強烈堅持先到先得的話，那么你就需要在構造重入鎖的時候，指定這是一個公平鎖：

ReentrantLock fairLock = new ReentrantLock(true); 

這樣一來，每一個請求鎖的線程，都會乖乖的把自己放入請求隊列，而不是上來就進行爭搶。但一定要注意，公平鎖是有代價的。維持公平競爭是以犧牲系統性能為代價的。如果你愿意承擔這個損失，公平鎖至少提供了一種普世價值觀的實現吧!

那公平鎖和非公平鎖實現的核心區別在哪里呢?來看一下這段lock()的代碼：

//非公平鎖  
 final void lock() { 
     //上來不管三七二十一，直接搶了再說 
     if (compareAndSetState(0, 1)) 
         setExclusiveOwnerThread(Thread.currentThread()); 
     else 
         //搶不到，就進隊列慢慢等著 
         acquire(1); 
 } 
 
 //公平鎖 
 final void lock() { 
     //直接進隊列等著 
     acquire(1); 
 } 

從上面的代碼中也不難看到，非公平鎖如果第一次爭搶失敗，后面的處理和公平鎖是一樣的，都是進入等待隊列慢慢等。

對于tryLock()也是非常類似的：

//非公平鎖  
final boolean nonfairTryAcquire(int acquires) { 
     final Thread current = Thread.currentThread(); 
     int c = getState(); 
     if (c == 0) { 
         //上來不管三七二十一，直接搶了再說 
         if (compareAndSetState(0, acquires)) { 
             setExclusiveOwnerThread(current); 
             return true; 
         } 
     } 
     //如果就是當前線程占用了鎖，那么就更新一下state，表示重復占用鎖的次數 
     //這是“重入”的關鍵所在 
     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 tryAcquire(int acquires) { 
    final Thread current = Thread.currentThread(); 
    int c = getState(); 
    if (c == 0) { 
        //先看看有沒有別人在等，沒有人等我才會去搶，有人在我前面 ，我就不搶啦 
        if (!hasQueuedPredecessors() && 
            compareAndSetState(0, acquires)) { 
            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; 
} 

Condition

Condition可以理解為重入鎖的伴生對象。它提供了在重入鎖的基礎上，進行等待和通知的機制?？梢允褂? newCondition()方法生成一個Condition對象，如下所示。

private final Lock lock = new ReentrantLock(); 
private final Condition condition = lock.newCondition(); 

那Condition對象怎么用呢。在JDK內部就有一個很好的例子。讓我們來看一下ArrayBlockingQueue吧。ArrayBlockingQueue是一個隊列，你可以把元素塞入隊列(enqueue)，也可以拿出來take()。但是有一個小小的條件，就是如果隊列是空的，那么take()就需要等待，一直等到有元素了，再返回。那這個功能，怎么實現呢?這就可以使用Condition對象了。

實現在ArrayBlockingQueue中，就維護一個Condition對象

lock = new ReentrantLock(fair); 
notEmpty = lock.newCondition(); 

這個notEmpty 就是一個Condition對象。它用來通知其他線程，ArrayBlockingQueue是不是空著的。當我們需要拿出一個元素時:

public E take() throws InterruptedException { 
    final ReentrantLock lock = this.lock; 
    lock.lockInterruptibly(); 
    try { 
        while (count == 0) 
            // 如果隊列長度為0，那么就在notEmpty condition上等待了，一直等到有元素進來為止 
            // 注意，await()方法，一定是要先獲得condition伴生的那個lock，才能用的哦 
            notEmpty.await(); 
        //一旦有人通知我隊列里有東西了，我就彈出一個返回 
        return dequeue(); 
    } finally { 
        lock.unlock(); 
    } 
} 

當有元素入隊時：

 public boolean offer(E e) { 
    checkNotNull(e); 
    final ReentrantLock lock = this.lock; 
    //先拿到鎖，拿到鎖才能操作對應的Condition對象 
    lock.lock(); 
    try { 
        if (count == items.length) 
            return false; 
        else { 
            //入隊了， 在這個函數里，就會進行notEmpty的通知，通知相關線程，有數據準備好了 
            enqueue(e); 
            return true; 
        } 
    } finally { 
        //釋放鎖了，等著的那個線程，現在可以去彈出一個元素試試了 
        lock.unlock(); 
    } 
} 
 
private void enqueue(E x) { 
    final Object[] items = this.items; 
    items[putIndex] = x; 
    if (++putIndex == items.length) 
        putIndex = 0; 
    count++; 
    //元素已經放好了，通知那個等著拿東西的人吧 
    notEmpty.signal(); 
} 

因此，整個流程如圖所示：

 

重入鎖的使用示例

為了讓大家更好的理解重入鎖的使用方法?，F在我們使用重入鎖，實現一個簡單的計數器。這個計數器可以保證在多線程環境中，統計數據的精確性，請看下面示例代碼：

public class Counter { 
    //重入鎖 
    private final Lock lock = new ReentrantLock(); 
    private int count; 
    public void incr() { 
        // 訪問count時，需要加鎖 
        lock.lock(); 
        try { 
            count++; 
        } finally { 
            lock.unlock(); 
        } 
    } 
     
    public int getCount() { 
        //讀取數據也需要加鎖，才能保證數據的可見性 
        lock.lock(); 
        try { 
            return count; 
        }finally { 
            lock.unlock(); 
        } 
    } 
} 

總結

可重入鎖算是多線程的入門級別知識點，所以我把他當做多線程系列的第一章節，對于重入鎖，我們需要特別知道幾點：

1. 對于同一個線程，重入鎖允許你反復獲得通一把鎖，但是，申請和釋放鎖的次數必須一致。
2. 默認情況下，重入鎖是非公平的，公平的重入鎖性能差于非公平鎖
3. 重入鎖的內部實現是基于CAS操作的。
4. 重入鎖的伴生對象Condition提供了await()和singal()的功能，可以用于線程間消息通信。