Java高并發連載23-基于AQS實作自定義同步器

一、AQS-條件變數的支持（二）

• 在如下代碼中，當另外一個執行緒呼叫條件變數的signal方法的時候（必須先呼叫鎖的lock方法獲取鎖），在內部會把條件佇列里面隊頭的一個執行緒節點從條件佇列里面移除并且放入AQS的阻塞佇列里面，然后激活這個執行緒，

public final void signal() {
 if(!isHeldExclusively()) {
  throw IllegalMonitorException();
 }
 Node first = firstWaiter;
 if(first != null){
  // 將條件佇列頭元素移動到AQS佇列
  doSignal(first);
 }
}

• 需要注意的是，AQS提供了ConditionObject的實作，并沒有提供newCondition函式，該函式用來new一個ConditionObject物件，需要由AQS的子類來提供newConditon函式
• 下面來看當一個執行緒呼叫條件變數的await()方法而被阻塞后，如何將其放入條件佇列

private Node addConditionWaiter() {
 Node t = lastWaiter;
 ...
 // (1)
 Node node = new Node(Thread.currentThread(),Node.CONDITION);
 // (2)
 if(t == null){
  firstWaiter = node;
 }else {
  t.nextWaiter = node; // (3)
 }
 lastWaiter = node; // (4)
 return node;
}

• 代碼（1）首先根據根據當前執行緒創建了一個型別為Node.CONDITION的節點，然后通過代碼（2），（3），（4）在單向佇列尾部插入一個元素
• 注意：當多個執行緒同時呼叫lock.lock()方法獲取鎖時，只有一個執行緒獲取到了鎖，其他執行緒會被轉換為Node節點插入到lock鎖對應的AQS阻塞里面，并且做自旋CAS嘗試獲取鎖
• 如果獲取到了鎖的執行緒又呼叫對應條件變數的await()方法，則該執行緒會釋放獲取到的鎖，并被轉化為Node節點插入到條件變數對應的條件佇列里面
• 這時候因為呼叫lock.lock()方法被阻塞到AQS佇列里面的一個執行緒會獲取到被釋放的鎖，如果該執行緒也呼叫了條件變數的await()方法則該執行緒也會被放入條件變數的條件佇列里面
• 當另外一個執行緒呼叫條件變數的signal()或者signalAll()方法的時候，會把條件佇列里面的一個或者全部Node節點移動到AQS的阻塞佇列里面，等待時機獲取鎖，
• 最后使用一個圖總結：一個鎖對應一個AQS阻塞佇列，對應多個條件變數，每個條件變數有自己的一個條件佇列，
• 

二、基于AQS實作自定義同步器

• 基于AQS實作一個不可重入的鎖，自定義AQS需要重寫一系列的函式，還需要定義原子變數state的含義，在這里我們定義state為0表示目前鎖沒有被執行緒持有，state為1表示所已經被某一個執行緒持有，由于是不可重入鎖，所以不需要記錄持有鎖的執行緒獲取鎖的次數，另外，我們自定義的鎖支持條件變數，
• 下面來看一下代碼實作

package com.ruigege.LockSourceAnalysis6;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class NonReentrantLockME implements Lock,java.io.Serializable{
 // 內部幫助類
 private static class Sync extends AbstractQueueSynchronizer {
  // 是否鎖已經被持有
  protected boolean isHeldExclusively() {
   return getState() == 1;
  }
  
  // 如果state為0，則嘗試獲取鎖
  public boolean tryAcquire(int acquires) {
   assert acquires == 1;
   if(compareAndSetState(0,1)) {
    setExclusiveOwnerThread(Thread.currentThread());
    return true;
   }
   return false;
  }
  
  // 嘗試釋放鎖，設定state為0
  protected boolean tryRelease(int release) {
   assert releases == 1;
   if(getState() == 0) {
    throw new IllegalMonitorStateException();
   }
   setExclusiveOwnerThread(null);
   setState(0);
   return true;
  }
  
  // 提供條件變數介面
  Condition newConditon() {
   return new ConditionObject();
  }
 }
 
 // 創建一個Sync來做具體的作業
 private final Sync sync = new Sync();
 
 public void lock() {
  sync.acquire(1);
 }
 
 public boolean tryLock() {
  return sync.tryAcquire(1);
 }
 
 public void unlock() {
  sync.release(1);
  
 }
 public Condition newCondition() {
  return sync.newConditon();
 }
 
 public boolean isLocked() {
  return sync.isHeldExclusively();
 }
 
 public void lockInterruptibly() throws InterruptedException {
  sync.acquireInterruptibly(1);
 }

 public boolean tryLock(long timeout,TimeUnit unit) throws InterruptedException {
  return sync.tryAcquireNanos(1,unit.toNanos(timeout));
 }
}

• 如上面的代碼，NonReentrantLock定義了一個內部類Sync用來實作具體的鎖的操作，Sync則繼承了AQS ，由于我們實作的獨占模式的鎖，所以Sync重寫了tryAcquire\tryRelease和isHeldExclusively3個方法，另外Sync提供了newCondition這個方法用來支持條件變數，

三、原始碼：

• 所在包：com.ruigege.ConcurrentListSouceCodeAnalysis5
• https://github.com/ruigege66/ConcurrentJava
• CSDN：https://blog.csdn.net/weixin_44630050
• 博客園：https://www.cnblogs.com/ruigege0000/
• 歡迎關注微信公眾號：傅里葉變換，個人賬號，僅用于技術交流