快捷導(dǎo)航

Java 基于AQS實(shí)現(xiàn)自定義同步器的示例

更新時(shí)間：2021年03月01日 09:13:07 作者：心悅君兮君不知-睿

這篇文章主要介紹了Java 基于AQS實(shí)現(xiàn)自定義同步器的示例，幫助大家更好的理解和學(xué)習(xí)使用Java，感興趣的朋友可以了解下

一、AQS-條件變量的支持

在如下代碼中，當(dāng)另外一個(gè)線程調(diào)用條件變量的signal方法的時(shí)候（必須先調(diào)用鎖的lock方法獲取鎖），在內(nèi)部會(huì)把條件隊(duì)列里面隊(duì)頭的一個(gè)線程節(jié)點(diǎn)從條件隊(duì)列里面移除并且放入AQS的阻塞隊(duì)列里面，然后激活這個(gè)線程。

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

需要注意的是，AQS提供了ConditionObject的實(shí)現(xiàn)，并沒(méi)有提供newCondition函數(shù)，該函數(shù)用來(lái)new一個(gè)ConditionObject對(duì)象，需要由AQS的子類(lèi)來(lái)提供newConditon函數(shù)
下面來(lái)看當(dāng)一個(gè)線程調(diào)用條件變量的await()方法而被阻塞后，如何將其放入條件隊(duì)列

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）首先根據(jù)根據(jù)當(dāng)前線程創(chuàng)建了一個(gè)類(lèi)型為Node.CONDITION的節(jié)點(diǎn)，然后通過(guò)代碼（2），（3），（4）在單向隊(duì)列尾部插入一個(gè)元素
注意：當(dāng)多個(gè)線程同時(shí)調(diào)用lock.lock()方法獲取鎖時(shí)，只有一個(gè)線程獲取到了鎖，其他線程會(huì)被轉(zhuǎn)換為Node節(jié)點(diǎn)插入到lock鎖對(duì)應(yīng)的AQS阻塞里面，并且做自旋CAS嘗試獲取鎖
如果獲取到了鎖的線程又調(diào)用對(duì)應(yīng)條件變量的await()方法，則該線程會(huì)釋放獲取到的鎖，并被轉(zhuǎn)化為Node節(jié)點(diǎn)插入到條件變量對(duì)應(yīng)的條件隊(duì)列里面
這時(shí)候因?yàn)檎{(diào)用lock.lock()方法被阻塞到AQS隊(duì)列里面的一個(gè)線程會(huì)獲取到被釋放的鎖，如果該線程也調(diào)用了條件變量的await()方法則該線程也會(huì)被放入條件變量的條件隊(duì)列里面
當(dāng)另外一個(gè)線程調(diào)用條件變量的signal()或者signalAll()方法的時(shí)候，會(huì)把條件隊(duì)列里面的一個(gè)或者全部Node節(jié)點(diǎn)移動(dòng)到AQS的阻塞隊(duì)列里面，等待時(shí)機(jī)獲取鎖。
最后使用一個(gè)圖總結(jié)：一個(gè)鎖對(duì)應(yīng)一個(gè)AQS阻塞隊(duì)列，對(duì)應(yīng)多個(gè)條件變量，每個(gè)條件變量有自己的一個(gè)條件隊(duì)列。

二、基于AQS實(shí)現(xiàn)自定義同步器

基于AQS實(shí)現(xiàn)一個(gè)不可重入的鎖，自定義AQS需要重寫(xiě)一系列的函數(shù)，還需要定義原子變量state的含義，在這里我們定義state為0表示目前鎖沒(méi)有被線程持有，state為1表示所已經(jīng)被某一個(gè)線程持有，由于是不可重入鎖，所以不需要記錄持有鎖的線程獲取鎖的次數(shù)，另外，我們自定義的鎖支持條件變量。
下面來(lái)看一下代碼實(shí)現(xiàn)

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{
 // 內(nèi)部幫助類(lèi)
 private static class Sync extends AbstractQueueSynchronizer {
  // 是否鎖已經(jīng)被持有
  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;
  }
  
  // 嘗試釋放鎖，設(shè)置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();
  }
 }
 
 // 創(chuàng)建一個(gè)Sync來(lái)做具體的工作
 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定義了一個(gè)內(nèi)部類(lèi)Sync用來(lái)實(shí)現(xiàn)具體的鎖的操作，Sync則繼承了AQS ，由于我們實(shí)現(xiàn)的獨(dú)占模式的鎖，所以Sync重寫(xiě)了tryAcquire\tryRelease和isHeldExclusively3個(gè)方法，另外Sync提供了newCondition這個(gè)方法用來(lái)支持條件變量。