Java并發編程之顯式鎖機制詳解

        我們之前介紹過synchronized關鍵字實現程序的原子性操作，它的內部也是一種加鎖和解鎖機制，是一種聲明式的編程方式，我們只需要對方法或者代碼塊進行聲明，Java內部幫我們在調用方法之前和結束時加鎖和解鎖。而我們本篇將要介紹的顯式鎖是一種手動式的實現方式，程序員控制鎖的具體實現，雖然現在越來越趨向于使用synchronized直接實現原子操作，但是了解了Lock接口的具體實現機制將有助于我們對synchronized的使用。本文主要涉及以下一些內容：

• 接口Lock的基本組成成員
• 可重入鎖ReentrantLock的基本使用
• 深入ReentrantLock的實現原理
  

一、接口Lock的基本組成成員

Lock 位于java.util.concurrent.locks包下，源碼如下：

public interface Lock {
  void lock();
  void lockInterruptibly()
  boolean tryLock();
  boolean tryLock(long time, TimeUnit unit)
  void unlock();
  Condition newCondition();
}

其中，

void lock();：調用該方法將獲得一個鎖的入口
lockInterruptibly()：該方法也是去獲得一個鎖，但是它是響應中斷的，一旦在獲取的過程中遭遇中斷將拋出 InterruptedException。
boolean tryLock();：該方法嘗試著去獲得一個鎖，如果獲取失敗將返回false，并不會阻塞當前線程
boolean tryLock(long time, TimeUnit unit)：嘗試著去獲取一個鎖，如果獲取失敗，將阻塞等待指定的時間，期間如果能夠獲得鎖將返回true，否則返回false，響應中斷請求。
void unlock();：釋放一個鎖
Condition newCondition();：條件變量，留待下篇文章學習

二、可重入鎖ReentrantLock的基本使用

     ReentrantLock是接口 Lock的一個最主要的實現類，不僅實現了Lock中的基本的加鎖釋放鎖的方法，還擴展了自己的方法。它有兩個構造方法：

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

參數 fair用于保證鎖機制的公平策略，公平的策略會是的等待時間越長的線程優先獲得鎖。保證公平必然會降低性能，所以ReentrantLock默認并不保證公平。我們用ReentrantLock來實現對程序的原子操作：

public class MyThread extends Thread{
  
  private static Lock lock = new ReentrantLock();
  public static int count;
  
  @Override
  public void run() {
    try {
      Thread.sleep((int)Math.random()*100);
      lock.lock();
      count++;
      lock.unlock();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

當我們在主程序中啟動一百個線程隨機喚醒對count進行加一時，無論運行多少次，結果都是一百，也就是說我們的ReentrantLock是可以為我們保證原子操作的。

ReentrantLock還有一個特性就是可以重入性，即在本身獲得某個鎖的前提下可以隨意進入被該鎖鎖住的其他方法，對于一個鎖可以重復進入。除此之外，ReentrantLock還具有一些其他的有關鎖信息的方法：

• public int getHoldCount()：表示當前線程持有該鎖的數量
• public boolean isHeldByCurrentThread()：判斷鎖是否為當前線程持有
• public boolean isLocked()：判斷鎖是否為任意一個線程持有，如果有則返回true，否則返回false
• public final boolean hasQueuedThreads()：判斷該鎖上是否有線程進行等待
• public final int getQueueLength()：返回當前等待隊列的長度，也就是等待進入該鎖的線程個數
  

三、深入ReentrantLock的實現原理

ReentrantLock依賴CAS和LockSupport來實現，LockSupport有點像工具類，它主要提供兩類方法，park和unpark。

• public static void park()
• public static void parkNanos(long nanos)
• public static void parkUntil(long deadline)
• public static void unpark(Thread thread)
  

調用park方法會使得當前線程丟失CPU使用權，從Runnable狀態轉變為Waiting狀態。而unpark方法則反過來讓Waiting狀態的某個線程轉變狀態為Runnable，等待操作系統調度。parkNanos和parkUntil是和時間相關的兩個park的變種，parkNanos指定線程要等待的時間，parkUntil則指定線程要等待到什么時候，這個時間是一個絕對時間，相對于紀元的毫秒數。

Java的并發包中有很多并發工具，ReentrantReadWriteLock，Semaphore，CountDownLatch，ReentrantLock等。這些工具有很多的共同特性，于是Java為我們抽象了一個類AbstractQueuedSynchronizer（AQS）來表示這些工具的共性。ReentrantLock是其的一個實現類，內部有三個內部類：

abstract static class Sync extends AbstractQueuedSynchronizer{
  //......
}

static final class NonfairSync extends Sync{
  //...........
}

static final class FairSync extends Sync {
  //.............
}

Sync 繼承了AQS并對其中的大部分代碼進行了簡單的實現，FairSync 和NonfairSync 是針對公平策略而定義的，如果構造ReentrantLock的時候指定公平的策略，那么其內部的所有方法都依賴這個FairSync ，否則就全部依賴NonfairSync。接著看ReentrantLock的構造函數：

private final Sync sync;

public ReentrantLock() {
  sync = new NonfairSync();
}

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

兩個構造方法最終會對sync進行初始化，而sync的將在后續的方法中起到相當大的作用。我們先看lock方法的具體實現：

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

ReentrantLock的lock方法調用的sync的lock方法，而在sync中的lock方法是一個抽象的方法，也就是說這個方法的具體實現在子類中，我們看NonfairSync中的實現：

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

AQS中有一個整型類型的State變量，它用于標識當前鎖被持有的次數，該值為0表示當前鎖沒有被任何線程持有。compareAndSetState是AQS中的方法，該方法調用了unsafe.compareAndSwapInt方法以CAS方式對State進行了更新，如果state的值為0，說明該鎖并沒有被任何線程持有，那么當前線程將持有該鎖并將state的值賦為1。

這就完成了獲取的動作，一旦后續的線程嘗試訪問臨界區代碼，在前面的線程沒有釋放鎖之前，將會調用 acquire(1)。

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

tryAcquire還是調用了AQS中的實現，

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;
}

第一個if判斷，想要持有的鎖是否被持有（雖然之前判斷過了，但是有可能在我們調用nonfairTryAcquire方法的期間，之前的線程釋放了該鎖），如果未被任何線程持有，那么將直接持有該鎖。

第二個if判斷，如果當前鎖的持有者就是當前線程，表示這是同線程的重入操作，于是增加鎖定次數并設置state的值。

整個方法結束之后，如果當前線程獲得了鎖，都將返回true，否則都會返回false。而如果tryAcquire方法返回true，那么整個acquire方法也將結束，否則就說明當前線程并沒有通過鎖，需要被阻塞。那么就會調用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法。

private Node addWaiter(Node mode) {
  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;
    }
  }
  enq(node);
  return node;
}

addWaiter方法將當前線程包裹成一個Node結點，添加到AQS內部所維護的一個等待隊列并返回該Node結點。最后調用acquireQueued方法：

final boolean acquireQueued(final Node node, int arg) {
  boolean failed = true;
  try {
    boolean interrupted = false;
    for (;;) {
      final Node p = node.predecessor();
      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);
  }
}

該方法首先會去獲得node的前一個結點，判斷如果是head結點，那么說明當前的node結點是整個等待隊列上的第一個等待的結點。于是讓它嘗試著去獲得鎖，如果能夠獲得鎖，將從等待隊列中清除它并返回。

如果發現當前結點前面還有等待的結點或者嘗試獲取鎖失敗，那么將會調用shouldParkAfterFailedAcquire方法判斷該結點鎖對應的線程是否需要被unpark阻塞，并最終調用LockSupport.park(this)阻塞當前線程。

在第一個線程持有該鎖的前提下，成功阻塞了第二個線程。這大概就是整個lock方法的調用鏈流程。

接下來看看unlock的具體實現，

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

這是ReentrantLock中對AQS的unlock的具體實現，調用了sync的release方法，這個方法是其父類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;
}

tryRelease被sync重寫，具體代碼如下：

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;
}

首先判斷如果當前線程并不是鎖的當前持有者，拋出異常（不持有該鎖自然不能釋放該鎖）。如果c等于0則表示，當前鎖只被持有一次，也就是當前線程并沒有多次重入該鎖，于是將該鎖的持有者設置為null，表示未被任何線程持有。如果c不等于0，那么說明該鎖被當前線程重入多次，于是對state減一并設置state的值。最終如果返回true則說明該鎖被釋放了，否則說明當前線程依然持有該鎖。

回到release方法，如果tryRelease(arg)返回true，那么方法體會判斷當前等待隊列是否有結點在等待該鎖，如果有則調用unparkSuccessor(h)方法喚醒等待隊列上的第一個等待的結點線程并返回true。

這里有一個細節，其實所有未能獲得鎖的線程都被阻塞在方法中：

final boolean acquireQueued(final Node node, int arg) {
  boolean failed = true;
  try {
    boolean interrupted = false;
    for (;;) {
      final Node p = node.predecessor();
      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);
  }
}

未能獲得鎖的線程被方法parkAndCheckInterrupt阻塞了，所以當我們在unlock中調用unpark喚醒一個等待隊列上的線程結點時，線程將從此處重新進入死循環嘗試去獲取鎖。如果能夠獲得鎖，將從等待隊列中移除自己，并返回，否則再次被阻塞等待喚醒。

整個unlock方法的執行流程也已經大致介紹完成，最后我們看看可重入鎖ReentrantLock和synchronized的一些對比。

四、ReentrantLock對比synchronized

     synchronized更傾向于一種聲明式的編程方式，我們在方法前使用synchronized修飾，Java會自動為我們實現其內部的細節，什么時候加鎖，什么時候釋放鎖都是它負責的。
     而對于我們的ReentrantLock重入鎖來說，需要我們自己手動的去加鎖和釋放鎖，對于邏輯的要求更高，也相對更難。
     而隨著jvm版本的更新和優化，ReentrantLock和synchronized在性能上的差別在逐漸縮小，所以一般建議使用synchronized而盡量避免復雜難操作的ReentrantLock。

對于顯式鎖的基本情況大致介紹如上，如有錯誤之處，望指出！

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持億速云。