Java 基于AQS实现自定义同步器的示例

时间：2022-03-25 17:43:35

一、AQS-条件变量的支持

在如下代码中，当另外一个线程调用条件变量的signal方法的时候（必须先调用锁的lock方法获取锁），在内部会把条件队列里面队头的一个线程节点从条件队列里面移除并且放入AQS的阻塞队列里面，然后激活这个线程。

  1. public final void signal() {
  2.  if(!isHeldExclusively()) {
  3.   throw IllegalMonitorException();
  4.  }
  5.  Node first = firstWaiter;
  6.  if(first != null){
  7.   // 将条件队列头元素移动到AQS队列
  8.   doSignal(first);
  9.  }
  10. }
  • 需要注意的是，AQS提供了ConditionObject的实现，并没有提供newCondition函数，该函数用来new一个ConditionObject对象，需要由AQS的子类来提供newConditon函数
  • 下面来看当一个线程调用条件变量的await()方法而被阻塞后，如何将其放入条件队列
  1. private Node addConditionWaiter() {
  2.  Node t = lastWaiter;
  3.  ...
  4.  // (1)
  5.  Node node = new Node(Thread.currentThread(),Node.CONDITION);
  6.  // (2)
  7.  if(t == null){
  8.   firstWaiter = node;
  9.  }else {
  10.   t.nextWaiter = node; // (3)
  11.  }
  12.  lastWaiter = node; // (4)
  13.  return node;
  14. }
  • 代码（1）首先根据根据当前线程创建了一个类型为Node.CONDITION的节点，然后通过代码（2），（3），（4）在单向队列尾部插入一个元素
  • 注意：当多个线程同时调用lock.lock()方法获取锁时，只有一个线程获取到了锁，其他线程会被转换为Node节点插入到lock锁对应的AQS阻塞里面，并且做自旋CAS尝试获取锁
  • 如果获取到了锁的线程又调用对应条件变量的await()方法，则该线程会释放获取到的锁，并被转化为Node节点插入到条件变量对应的条件队列里面
  • 这时候因为调用lock.lock()方法被阻塞到AQS队列里面的一个线程会获取到被释放的锁，如果该线程也调用了条件变量的await()方法则该线程也会被放入条件变量的条件队列里面
  • 当另外一个线程调用条件变量的signal()或者signalAll()方法的时候，会把条件队列里面的一个或者全部Node节点移动到AQS的阻塞队列里面，等待时机获取锁。
  • 最后使用一个图总结：一个锁对应一个AQS阻塞队列，对应多个条件变量，每个条件变量有自己的一个条件队列。

Java 基于AQS实现自定义同步器的示例

二、基于AQS实现自定义同步器

  • 基于AQS实现一个不可重入的锁，自定义AQS需要重写一系列的函数，还需要定义原子变量state的含义，在这里我们定义state为0表示目前锁没有被线程持有，state为1表示所已经被某一个线程持有，由于是不可重入锁，所以不需要记录持有锁的线程获取锁的次数，另外，我们自定义的锁支持条件变量。
  • 下面来看一下代码实现
  1. package com.ruigege.LockSourceAnalysis6;
  2.  
  3. import java.util.concurrent.TimeUnit;
  4. import java.util.concurrent.locks.Condition;
  5. import java.util.concurrent.locks.Lock;
  6.  
  7. public class NonReentrantLockME implements Lock,java.io.Serializable{
  8.  // 内部帮助类
  9.  private static class Sync extends AbstractQueueSynchronizer {
  10.   // 是否锁已经被持有
  11.   protected boolean isHeldExclusively() {
  12.    return getState() == 1;
  13.   }
  14.   
  15.   // 如果state为0，则尝试获取锁
  16.   public boolean tryAcquire(int acquires) {
  17.    assert acquires == 1;
  18.    if(compareAndSetState(0,1)) {
  19.     setExclusiveOwnerThread(Thread.currentThread());
  20.     return true;
  21.    }
  22.    return false;
  23.   }
  24.   
  25.   // 尝试释放锁，设置state为0
  26.   protected boolean tryRelease(int release) {
  27.    assert releases == 1;
  28.    if(getState() == 0) {
  29.     throw new IllegalMonitorStateException();
  30.    }
  31.    setExclusiveOwnerThread(null);
  32.    setState(0);
  33.    return true;
  34.   }
  35.   
  36.   // 提供条件变量接口
  37.   Condition newConditon() {
  38.    return new ConditionObject();
  39.   }
  40.  }
  41.  
  42.  // 创建一个Sync来做具体的工作
  43.  private final Sync sync = new Sync();
  44.  
  45.  public void lock() {
  46.   sync.acquire(1);
  47.  }
  48.  
  49.  public boolean tryLock() {
  50.   return sync.tryAcquire(1);
  51.  }
  52.  
  53.  public void unlock() {
  54.   sync.release(1);
  55.   
  56.  }
  57.  public Condition newCondition() {
  58.   return sync.newConditon();
  59.  }
  60.  
  61.  public boolean isLocked() {
  62.   return sync.isHeldExclusively();
  63.  }
  64.  
  65.  public void lockInterruptibly() throws InterruptedException {
  66.   sync.acquireInterruptibly(1);
  67.  }
  68.  
  69.  public boolean tryLock(long timeout,TimeUnit unit) throws InterruptedException {
  70.   return sync.tryAcquireNanos(1,unit.toNanos(timeout));
  71.  }
  72. }

如上面的代码，NonReentrantLock定义了一个内部类Sync用来实现具体的锁的操作，Sync则继承了AQS ，由于我们实现的独占模式的锁，所以Sync重写了tryAcquire\tryRelease和isHeldExclusively3个方法，另外Sync提供了newCondition这个方法用来支持条件变量。

三、源码：

所在包：com.ruigege.ConcurrentListSouceCodeAnalysis5

https://github.com/ruigege66/ConcurrentJava

以上就是Java 基于AQS实现自定义同步器的示例的详细内容，更多关于Java 基于AQS实现自定义同步器的资料请关注服务器之家其它相关文章！

原文链接：https://www.cnblogs.com/ruigege0000/p/14450174.html

标签：

相关文章