Java基礎——深入理解ReentrantLock

一、簡介

? ? ? ?在Java中通常實現鎖有兩種方式，一種是synchronized關鍵字，另一種是Lock。二者其實并沒有什么必然聯系，但是各有各的特點，在使用中可以進行取舍的使用。

二、ReentrantLock與synchronized的比較

相同點：

（1）ReentrantLock提供了synchronized類似的功能和內存語義。

不同點：

? ? （1）ReentrantLock功能性方面更全面，比如定時等候鎖、可中斷鎖等候、鎖投票等，因此更有擴展性。在多個條件變量和高度競爭鎖的地方，用ReentrantLock更合適，ReentrantLock還提供了Condition，對線程的等待和喚醒等操作更加靈活，一個ReentrantLock可以有多個Condition實例，所以更有擴展性。

? ? （2）ReentrantLock 的性能比synchronized會好點。

? ? （3）ReentrantLock提供了可輪詢的鎖請求，他可以嘗試的去取得鎖，如果取得成功則繼續處理，取得不成功，可以等下次運行的時候處理，所以不容易產生死鎖，而synchronized則一旦進入鎖請求要么成功，要么一直阻塞，所以更容易產生死鎖。

? ? （4）對于使用者的直觀體驗上Lock是比較復雜的，需要lock和realse，通常需要在finally中進行鎖的釋放，否則，如果受保護的代碼將拋出異常，就會產生死鎖的問題，這一點區別看起來可能沒什么，但是實際上，它極為重要。但是synchronized的使用十分簡單，只需要對自己的方法或者關注的同步對象或類使用synchronized關鍵字即可，JVM 將確保鎖會獲得自動釋放。但是對于鎖的粒度控制比較粗，同時對于實現一些鎖的狀態的轉移比較困難。

? ? （5）當 JVM 用 synchronized 管理鎖定請求和釋放時，JVM 在生成線程轉儲時能夠包括鎖定信息。這些對調試非常有價值，因為它們能標識死鎖或者其他異常行為的來源。 Lock 類只是普通的類，JVM 不知道具體哪個線程擁有 Lock 對象。

三、ReentrantLock

? ? ? ?java.util.concurrent.lock 中的 Lock 框架是鎖定的一個抽象，它允許把鎖定的實現作為 Java 類，而不是作為語言的特性來實現。這就為 Lock 的多種實現留下了空間，各種實現可能有不同的調度算法、性能特性或者鎖定語義。 ReentrantLock 類實現了 Lock ，它擁有與 synchronized 相同的并發性和內存語義，但是添加了類似鎖投票、定時鎖等候和可中斷鎖等候的一些特性。此外，它還提供了在激烈爭用情況下更佳的性能。（換句話說，當許多線程都想訪問共享資源時，JVM 可以花更少的時候來調度線程，把更多時間用在執行線程上。）

? ? ? ?reentrant 鎖意味著什么呢？簡單來說，它有一個與鎖相關的獲取計數器，如果擁有鎖的某個線程再次得到鎖，那么獲取計數器就加1，然后鎖需要被釋放兩次才能獲得真正釋放。這模仿了 synchronized 的語義；如果線程進入由線程已經擁有的監控器保護的 synchronized 塊，就允許線程繼續進行，當線程退出第二個（或者后續） synchronized 塊的時候，不釋放鎖，只有線程退出它進入的監控器保護的第一個 synchronized 塊時，才釋放鎖。

1、實現可輪詢的鎖請求

? ? ? ?在內部鎖中，死鎖是致命的——唯一的恢復方法是重新啟動程序，唯一的預防方法是在構建程序時不要出錯。而可輪詢的鎖獲取模式具有更完善的錯誤恢復機制，可以規避死鎖的發生。?
? ? ? ?如果你不能獲得所有需要的鎖，那么使用可輪詢的獲取方式使你能夠重新拿到控制權，它會釋放你已經獲得的這些鎖，然后再重新嘗試。可輪詢的鎖獲取模式，由tryLock()方法實現。此方法僅在調用時鎖為空閑狀態才獲取該鎖。如果鎖可用，則獲取鎖，并立即返回值true。如果鎖不可用，則此方法將立即返回值false。此方法的典型使用語句如下：?

Lock lock = ...;   
if (lock.tryLock()) {   try {   // manipulate protected state   } finally {   lock.unlock();   }   
} else {   // perform alternative actions   
}

2、實現可定時的鎖請求

? ? ? ?當使用內部鎖時，一旦開始請求，鎖就不能停止了，所以內部鎖給實現具有時限的活動帶來了風險。為了解決這一問題，可以使用定時鎖。當具有時限的活?動調用了阻塞方法，定時鎖能夠在時間預算內設定相應的超時。如果活動在期待的時間內沒能獲得結果，定時鎖能使程序提前返回。可定時的鎖獲取模式，由tryLock(long, TimeUnit)方法實現。?

3、實現可中斷的鎖獲取請求?
? ? ? ?可中斷的鎖獲取操作允許在可取消的活動中使用。lockInterruptibly()方法能夠使你獲得鎖的時候響應中斷。

四、條件變量Condition

? ? ? ?條件變量很大一個程度上是為了解決Object.wait/notify/notifyAll難以使用的問題。

? ? ? ?條件（也稱為條件隊列?或條件變量）為線程提供了一個含義，以便在某個狀態條件現在可能為 true 的另一個線程通知它之前，一直掛起該線程（即讓其“等待”）。因為訪問此共享狀態信息發生在不同的線程中，所以它必須受保護，因此要將某種形式的鎖與該條件相關聯。等待提供一個條件的主要屬性是：以原子方式?釋放相關的鎖，并掛起當前線程，就像?Object.wait?做的那樣。

? ? ? ?上述API說明表明條件變量需要與鎖綁定，而且多個Condition需要綁定到同一鎖上。前面的Lock中提到，獲取一個條件變量的方法是Lock.newCondition()。

void await() throws InterruptedException;  void awaitUninterruptibly();  long awaitNanos(long nanosTimeout) throws InterruptedException;  boolean await(long time, TimeUnit unit) throws InterruptedException;  boolean awaitUntil(Date deadline) throws InterruptedException;  void signal();  void signalAll();

? ? ? ?以上是Condition接口定義的方法，await*對應于Object.wait，signal對應于Object.notify，signalAll對應于Object.notifyAll。特別說明的是Condition的接口改變名稱就是為了避免與Object中的wait/notify/notifyAll的語義和使用上混淆，因為Condition同樣有wait/notify/notifyAll方法。

? ? ? ?每一個Lock可以有任意數據的Condition對象，Condition是與Lock綁定的，所以就有Lock的公平性特性：如果是公平鎖，線程為按照FIFO的順序從Condition.await中釋放，如果是非公平鎖，那么后續的鎖競爭就不保證FIFO順序了。

一個使用Condition實現生產者消費者的模型例子如下。

import java.util.concurrent.locks.Condition;  
import java.util.concurrent.locks.Lock;  
import java.util.concurrent.locks.ReentrantLock;  public class ProductQueue<T> {  private final T[] items;  private final Lock lock = new ReentrantLock();  private Condition notFull = lock.newCondition();  private Condition notEmpty = lock.newCondition();  //  private int head, tail, count;  public ProductQueue(int maxSize) {  items = (T[]) new Object[maxSize];  }  public ProductQueue() {  this(10);  }  public void put(T t) throws InterruptedException {  lock.lock();  try {  while (count == getCapacity()) {  notFull.await();  }  items[tail] = t;  if (++tail == getCapacity()) {  tail = 0;  }  ++count;  notEmpty.signalAll();  } finally {  lock.unlock();  }  }  public T take() throws InterruptedException {  lock.lock();  try {  while (count == 0) {  notEmpty.await();  }  T ret = items[head];  items[head] = null;//GC  //  if (++head == getCapacity()) {  head = 0;  }  --count;  notFull.signalAll();  return ret;  } finally {  lock.unlock();  }  }  public int getCapacity() {  return items.length;  }  public int size() {  lock.lock();  try {  return count;  } finally {  lock.unlock();  }  }  }

? ? ? ?在這個例子中消費take()需要隊列不為空，如果為空就掛起（await()），直到收到notEmpty的信號；生產put()需要隊列不滿，如果滿了就掛起（await()），直到收到notFull的信號。

? ? ? ?可能有人會問題，如果一個線程lock()對象后被掛起還沒有unlock，那么另外一個線程就拿不到鎖了（lock()操作會掛起），那么就無法通知(notify)前一個線程，這樣豈不是“死鎖”了？

1、await* 操作

? ? ? ?上一節中說過多次ReentrantLock是獨占鎖，一個線程拿到鎖后如果不釋放，那么另外一個線程肯定是拿不到鎖，所以在lock.lock()和lock.unlock()之間可能有一次釋放鎖的操作（同樣也必然還有一次獲取鎖的操作）。我們再回頭看代碼，不管take()還是put()，在進入lock.lock()后唯一可能釋放鎖的操作就是await()了。也就是說await()操作實際上就是釋放鎖，然后掛起線程，一旦條件滿足就被喚醒，再次獲取鎖！

public final void await() throws InterruptedException {  if (Thread.interrupted())  throw new InterruptedException();  Node node = addConditionWaiter();  int savedState = fullyRelease(node);  int interruptMode = 0;  while (!isOnSyncQueue(node)) {  LockSupport.park(this);  if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)  break;  }  if (acquireQueued(node, savedState) && interruptMode != THROW_IE)  interruptMode = REINTERRUPT;  if (node.nextWaiter != null)  unlinkCancelledWaiters();  if (interruptMode != 0)  reportInterruptAfterWait(interruptMode);  
}

? ? ? ?上面是await()的代碼片段。上一節中說過,AQS在獲取鎖的時候需要有一個CHL的FIFO隊列，所以對于一個Condition.await()而言，如果釋放了鎖，要想再一次獲取鎖那么就需要進入隊列，等待被通知獲取鎖。完整的await()操作是安裝如下步驟進行的：

將當前線程加入Condition鎖隊列。特別說明的是，這里不同于AQS的隊列，這里進入的是Condition的FIFO隊列。后面會具體談到此結構。進行2。
釋放鎖。這里可以看到將鎖釋放了，否則別的線程就無法拿到鎖而發生死鎖。進行3。
自旋(while)掛起，直到被喚醒或者超時或者CACELLED等。進行4。
獲取鎖(acquireQueued)。并將自己從Condition的FIFO隊列中釋放，表明自己不再需要鎖（我已經拿到鎖了）。

? ? ? ?這里再回頭介紹Condition的數據結構。我們知道一個Condition可以在多個地方被await*()，那么就需要一個FIFO的結構將這些Condition串聯起來，然后根據需要喚醒一個或者多個（通常是所有）。所以在Condition內部就需要一個FIFO的隊列。

private transient Node firstWaiter;  
private transient Node lastWaiter;

? ? ? ?上面的兩個節點就是描述一個FIFO的隊列。我們再結合前面提到的節點（Node）數據結構。我們就發現Node.nextWaiter就派上用場了！nextWaiter就是將一系列的Condition.await*串聯起來組成一個FIFO的隊列。

2、signal/signalAll 操作

? ? ? ?await*()清楚了，現在再來看signal/signalAll就容易多了。按照signal/signalAll的需求，就是要將Condition.await*()中FIFO隊列中第一個Node喚醒（或者全部Node）喚醒。盡管所有Node可能都被喚醒，但是要知道的是仍然只有一個線程能夠拿到鎖，其它沒有拿到鎖的線程仍然需要自旋等待，就上上面提到的第4步(acquireQueued)。

private void doSignal(Node first) {  do {  if ( (firstWaiter = first.nextWaiter) == null)  lastWaiter = null;  first.nextWaiter = null;  } while (!transferForSignal(first) &&  (first = firstWaiter) != null);  
}  private void doSignalAll(Node first) {  lastWaiter = firstWaiter  = null;  do {  Node next = first.nextWaiter;  first.nextWaiter = null;  transferForSignal(first);  first = next;  } while (first != null);  
}

? ? ? ?上面的代碼很容易看出來，signal就是喚醒Condition隊列中的第一個非CANCELLED節點線程，而signalAll就是喚醒所有非CANCELLED節點線程。當然了遇到CANCELLED線程就需要將其從FIFO隊列中剔除。

final boolean transferForSignal(Node node) {  if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))  return false;  Node p = enq(node);  int c = p.waitStatus;  if (c > 0 || !compareAndSetWaitStatus(p, c, Node.SIGNAL))  LockSupport.unpark(node.thread);  return true;  
}