AQS原理分析

什么是AQS

java.util.concurrent包中的大多数同步器实现都是围绕着共同的基础行为，比如等待队列、条件队列、独占获取、共享获取等，而这些行为的抽象就是基于AbstractQueuedSynchronizer（简称AQS）实现的，AQS是一个抽象同步框架，可以用来实现一个依赖状态的同步器。
AQS具备的特性：

  • 阻塞等待队列
  • 共享/独占
  • 公平/非公平
  • 可重入
  • 允许中断

AQS内部维护属性volatile int state

  • state表示资源的可用状态

State三种访问方式：

  • getState()
  • setState()
  • compareAndSetState()

AQS定义两种资源共享方式

  • Exclusive-独占，只有一个线程能执行，如ReentrantLock
  • Share-共享，多个线程可以同时执行，如Semaphore/CountDownLatch

AQS定义两种队列

  • 同步等待队列

  • 条件等待队列

    同步等待队列

    AQS当中的同步等待队列也称CLH队列，CLH队列是Craig、Landin、Hagersten三人发明的一种基于双向链表数据结构的队列，是FIFO先入先出线程等待队列，Java中的CLH队列是原CLH队列的一个变种,线程由原自旋机制改为阻塞机制。
    AQS 依赖CLH同步队列来完成同步状态的管理：

  • 当前线程如果获取同步状态失败时，AQS则会将当前线程已经等待状态等信息构造成一个节点（Node）并将其加入到CLH同步队列，同时会阻塞当前线程

  • 当同步状态释放时，会把首节点唤醒（公平锁），使其再次尝试获取同步状态。
  • 通过signal或signalAll将条件队列中的节点转移到同步队列。（由条件队列转化为同步队列）

image.png

条件等待队列

AQS中条件队列是使用单向列表保存的，用nextWaiter来连接:

  • 调用await方法阻塞线程；
  • 当前线程存在于同步队列的头结点，调用await方法进行阻塞（从同步队列转化到条件队列）

    Condition接口详解

    image.png
  1. 调用Condition#await方法会释放当前持有的锁，然后阻塞当前线程，同时向Condition队列尾部添加一个节点，所以调用Condition#await方法的时候必须持有锁。
  2. 调用Condition#signal方法会将Condition队列的首节点移动到阻塞队列尾部，然后唤醒因调用Condition#await方法而阻塞的线程(唤醒之后这个线程就可以去竞争锁了)，所以调用Condition#signal方法的时候必须持有锁，持有锁的线程唤醒被因调用Condition#await方法而阻塞的线程。

    ReentrantLock详解

    ReentrantLock是一种基于AQS框架的应用实现，是JDK中的一种线程并发访问的同步手段，它的功能类似于synchronized是一种互斥锁，可以保证线程安全。
    相对于 synchronized， ReentrantLock具备如下特点：
  • 可中断
  • 可以设置超时时间
  • 可以设置为公平锁
  • 支持多个条件变量
  • 与 synchronized一样，都支持可重入

顺便总结了几点synchronized和ReentrantLock的区别：

  • synchronized是JVM层次的锁实现，ReentrantLock是JDK层次的锁实现；
  • synchronized的锁状态是无法在代码中直接判断的，但是ReentrantLock可以通过ReentrantLock#isLocked判断；
  • synchronized是非公平锁，ReentrantLock是可以是公平也可以是非公平的；
  • synchronized是不可以被中断的，而ReentrantLock#lockInterruptibly方法是可以被中断的；
  • 在发生异常时synchronized会自动释放锁，而ReentrantLock需要开发者在finally块中显示释放锁；
  • ReentrantLock获取锁的形式有多种：如立即返回是否成功的tryLock(),以及等待指定时长的获取，更加灵活；
  • synchronized在特定的情况下对于已经在等待的线程是后来的线程先获得锁（回顾一下sychronized的唤醒策略），而ReentrantLock对于已经在等待的线程是先来的线程先获得锁；

    ReentrantLock源码分析

  1. ReentrantLock加锁解锁的逻辑
  2. 公平和非公平，可重入锁的实现
  3. 线程竞争锁失败入队阻塞逻辑和获取锁的线程释放锁唤醒阻塞线程竞争锁的逻辑实现 （ 设计的精髓：并发场景下入队和出队操作）

公平

  1.          final void lock() {
  2.             acquire(1);
  3.         }

非公平

  1.         final void lock() {
  2.             if (compareAndSetState(0, 1))
  3.                 setExclusiveOwnerThread(Thread.currentThread());
  4.             else
  5.                 acquire(1);
  6.         }
  1.     protected final void setExclusiveOwnerThread(Thread thread) {
  2.         exclusiveOwnerThread = thread;
  3.     }
  1.     public final void acquire(int arg) {
  2.         if (!tryAcquire(arg) &&
  3.             acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  4.             selfInterrupt();
  5.     }
  1.         final boolean nonfairTryAcquire(int acquires) {
  2.             final Thread current = Thread.currentThread();
  3.             int c = getState();
  4.             if (c == 0) {
  5.                 if (compareAndSetState(0, acquires)) {
  6.                     setExclusiveOwnerThread(current);
  7.                     return true;
  8.                 }
  9.             }
  10.             else if (current == getExclusiveOwnerThread()) {
  11.                 int nextc = c + acquires;
  12.                 if (nextc < 0) // overflow
  13.                     throw new Error("Maximum lock count exceeded");
  14.                 setState(nextc);
  15.                 return true;
  16.             }
  17.             return false;
  18.         }
  1.     private Node addWaiter(Node mode) {
  2.         Node node = new Node(Thread.currentThread(), mode);
  3.         // Try the fast path of enq; backup to full enq on failure
  4.         Node pred = tail;
  5.         if (pred != null) {
  6.             node.prev = pred;
  7.             if (compareAndSetTail(pred, node)) {
  8.                 pred.next = node;
  9.                 return node;
  10.             }
  11.         }
  12.         enq(node);
  13.         return node;
  14.     }
  1.     private Node enq(final Node node) {
  2.         for (;;) {
  3.             Node t = tail;
  4.             if (t == null) { // Must initialize
  5.                 if (compareAndSetHead(new Node()))
  6.                     tail = head;
  7.             } else {
  8.                 node.prev = t;
  9.                 if (compareAndSetTail(t, node)) {
  10.                     t.next = node;
  11.                     return t;
  12.                 }
  13.             }
  14.         }
  15.     }
  1.     final boolean acquireQueued(final Node node, int arg) {
  2.         boolean failed = true;
  3.         try {
  4.             boolean interrupted = false;
  5.             for (;;) {
  6.                 final Node p = node.predecessor();
  7.                 if (p == head && tryAcquire(arg)) {
  8.                     setHead(node);
  9.                     p.next = null; // help GC
  10.                     failed = false;
  11.                     return interrupted;
  12.                 }
  13.                 if (shouldParkAfterFailedAcquire(p, node) &&
  14.                     parkAndCheckInterrupt())
  15.                     interrupted = true;
  16.             }
  17.         } finally {
  18.             if (failed)
  19.                 cancelAcquire(node);
  20.         }
  21.     }

image.png

解锁

  1.         protected final boolean tryRelease(int releases) {
  2.             int c = getState() - releases;
  3.             if (Thread.currentThread() != getExclusiveOwnerThread())
  4.                 throw new IllegalMonitorStateException();
  5.             boolean free = false;
  6.             if (c == 0) {
  7.                 free = true;
  8.                 setExclusiveOwnerThread(null);
  9.             }
  10.             setState(c);
  11.             return free;
  12.         }