Java 队列同步器 AQS(AbstractQueuedSynchronizer)源码解析

客官°小女子只卖身不卖艺 2024-04-19 10:36 80阅读 0赞

#### 文章目录 ####

*  1. AQS 同步原理
 *   *   *  1.1 核心思想
         *  1.2 锁获取释放的实现
         *   *  1.2.1 获取锁过程
             *  1.2.2 释放锁过程
         *  1.3 AQS 的资源分配方式
 *  2. JDK 中 AQS 的应用
 *   *   *  2.1 ReentrantLock
         *  2.2 Semaphore
         *  2.3 CountDownLatch
         *  2.4 CyclicBarrier

## 1. AQS 同步原理 ##

#### 1.1 核心思想 ####

如我们所知，同步是为了解决共享资源的使用安全问题，其实现核心主要分为两个部分：

> 1.  **锁的语义实现**  
>     就是使用某种共享资源作为锁，线程通过争用这个资源从而获得执行权
> 2.  **线程调度**  
>     表现为锁争用失败后对线程的处理策略，以及锁释放后线程的唤醒和锁分配

AQS核心思想是，如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并且将共享资源设置为锁定状态。如果被请求的共享资源被占用，那就将暂时获取不到资源的线程封装成一个 Node 节点加入到队列中阻塞等待。当持有资源的线程释放资源时，会唤醒后继结点，唤醒的结点线程继续加入到对资源的争夺中

*  **AQS 锁的语义实现**：  
    使用一个 int 成员变量来表示同步状态，通过 CAS 对该同步状态进行原子操作实现对值的修改
    
        /**
          * The synchronization state.
          */
         private volatile int state;//共享变量，使用volatile修饰保证线程可见性
 *  **AQS 线程调度**  
    通过 FIFO 队列来完成获取资源线程的排队工作，这个机制在 AQS 中是用 CLH 队列（虚拟双向队列，不存在队列实例仅存在结点间的指向关系） 实现的，即将暂时获取不到资源的线程封装成一个 Node 节点加入到队列中
    
        To enqueue into a CLH lock, you atomically splice it in as new tail. To dequeue, you 
        just set the head field.
        
          *      +------+  prev +-----+  prev +-----+
          * head | Node | <---- | Node| <---- | Node|  tail
          *      |      | ----> |     | ----> |     |
          *      +------+ next  +-----+ next  +-----+
        
        static final class Node {
              
         	int waitStatus;
         	Node prev;
         	Node next;
         	Node nextWaiter;
         	Thread thread;
        }

#### 1.2 锁获取释放的实现 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70_pic_center]

##### 1.2.1 获取锁过程 #####

1.  AQS 使用 `AbstractQueuedSynchronizer#acquire()` 方法来尝试获取锁，对中断不敏感，完成 synchronized 语义，这部分大概包含了以下几个步骤：
    
    > 1.  调用子类重写的`tryAcquire()`方法尝试更改状态值（也就是尝试获取锁），该方法中通常会使用 AQS 提供的对 state 操作的方法，利用`CAS`保证只有一个线程能够对状态成功修改，而没有成功修改的线程也就意味着获取锁失败，将进入 sync 队列等待
    > 2.  如果获取不到锁，首先调用 `addWaiter()`将当前线程构造成节点 Node 并加入等待队列
    > 3.  `AbstractQueuedSynchronizer#acquireQueued()` 方法使节点在队列中空循环一直尝试获取锁，直到当前节点的前继节点成为 head 节点并且 `tryAcquire()`资源成功，则设置当前节点为 head 并退出等待队列；在空循环中如检测到需要阻塞当前线程，则调用 `parkAndCheckInterrupt()`将当前线程从线程调度器上摘下，进入等待状态
    
        public final void acquire(int arg) {
              
             if (!tryAcquire(arg) &&
                 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                 selfInterrupt();
         }
2.  `AbstractQueuedSynchronizer#addWaiter()` 方法实现比较简单，就是构造等待节点，每个新的 Node 节点入队都会使用 CAS 将其放在队列尾部，也就是将 `tail 尾部指针`指向它，从而形成了一个虚拟双向队列，这样做的目的是将线程间的通信限制在较小规模（也就是2个节点左右）
    
        private Node addWaiter(Node mode) {
              
             Node node = new Node(Thread.currentThread(), mode);
             // Try the fast path of enq; backup to full enq on failure
             Node pred = tail;
             if (pred != null) {
              
                 node.prev = pred;
                 if (compareAndSetTail(pred, node)) {
              
                     pred.next = node;
                     return node;
                 }
             }
             enq(node);
             return node;
         }
         
         private Node enq(final Node node) {
              
             for (;;) {
              
                 Node t = tail;
                 if (t == null) {
               // Must initialize
                     if (compareAndSetHead(new Node()))
                         tail = head;
                 } else {
              
                     node.prev = t;
                     if (compareAndSetTail(t, node)) {
              
                         t.next = node;
                         return t;
                     }
                 }
             }
         }
3.  `AbstractQueuedSynchronizer#acquireQueued()` 方法的实现也很简洁，需要注意的主要有以下几点：
    
    > 1.  首先可以看到 for 空循环中会检查当前节点的前继节点是否是等待队列中的 head 头节点，如果是的话再次调用 `tryAcquire()` 方法去尝试获取锁，获取成功则将头节点设置为当前节点，其实这一步操作也就是把当前节点从等待队列中移除了
    > 2.  以上条件不成立，如果检查到当前线程需要挂起，则执行挂起操作，线程执行在此中断。等到锁被释放时，释放锁的线程会从等待队列头节点位置唤醒在等待队列中的节点，则等待节点的线程从挂起处继续执行，在 for 空循环中重复以上步骤
    
        final boolean acquireQueued(final Node node, int arg) {
              
             boolean failed = true;
             try {
              
                 boolean interrupted = false;
                 for (;;) {
              
                     final Node p = node.predecessor();
                     if (p == head && tryAcquire(arg)) {
              
                         setHead(node);
                         p.next = null; // help GC
                         failed = false;
                         return interrupted;
                     }
                     // 在此处挂起，则唤醒后继续从此处执行
                     if (shouldParkAfterFailedAcquire(p, node) &&
                         parkAndCheckInterrupt())
                         interrupted = true;
                 }
             } finally {
              
                 if (failed)
                     cancelAcquire(node);
             }
         }
4.  `AbstractQueuedSynchronizer#parkAndCheckInterrupt()` 主要用于判断当前线程是否需要挂起，从其实现看主要逻辑如下：
    
    > 1.  首先判断当前等待节点的前继节点的状态是否是 `Node.SIGNAL`，是的话说明已经告知了前继节点需要在锁释放的时候唤醒当前节点，那么当前节点可以安全地挂起了
    > 2.  如果当前节点的前继节点所对应的状态大于 0，也就是被 cancel 了，则调整等待队列结构，将当前节点挂在离它最近的没有 cancel 的前继节点后面
    > 3.  以上条件都不满足，则将当前节点的前继节点的等待状态使用 CAS 设置为 `Node.SIGNAL`，也就意味着当时机合适的时候前继节点需要执行唤醒当前节点的操作
    
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
              
             int ws = pred.waitStatus;
             if (ws == Node.SIGNAL)
                 /*
                  * This node has already set status asking a release
                  * to signal it, so it can safely park.
                  */
                 return true;
             if (ws > 0) {
              
                 /*
                  * Predecessor was cancelled. Skip over predecessors and
                  * indicate retry.
                  */
                 do {
              
                     node.prev = pred = pred.prev;
                 } while (pred.waitStatus > 0);
                 pred.next = node;
             } else {
              
                 /*
                  * waitStatus must be 0 or PROPAGATE.  Indicate that we
                  * need a signal, but don't park yet.  Caller will need to
                  * retry to make sure it cannot acquire before parking.
                  */
                 compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
             }
             return false;
         }

##### 1.2.2 释放锁过程 #####

1.  AQS 使用 `AbstractQueuedSynchronizer#release()`方法将状态设置回去，也就是将锁释放，其主要逻辑包括：
    
    > 1.  尝试释放资源，`tryRelease()` 通过 `CAS`保证原子化的将状态设置回去，如果释放状态成功过则将进入将等待队列中线程唤醒的流程。此处需要校验等待队列头节点的等待状态，以便确定等待队列中是否真的存在等待唤醒的线程
    > 2.  `AbstractQueuedSynchronizer#unparkSuccessor()`尝试唤醒等待队列头节点的后继节点所包含的线程
    
        public final boolean release(int arg) {
              
             if (tryRelease(arg)) {
              
                 Node h = head;
                 if (h != null && h.waitStatus != 0)
                     unparkSuccessor(h);
                 return true;
             }
             return false;
         }
2.  `AbstractQueuedSynchronizer#unparkSuccessor()`方法的实现如下，可以看到首先会检查头节点的后继节点，如果该节点满足状态（没有被cancel掉）就进行唤醒操作，如果不满足状态，则从尾部开始找寻需要被唤醒的节点，之后通过LockSupport 的`unpark()`方法唤醒节点，使其从挂起处重新执行，继续尝试获取资源
    
        private void unparkSuccessor(Node node) {
              
             /*
              * If status is negative (i.e., possibly needing signal) try
              * to clear in anticipation of signalling.  It is OK if this
              * fails or if status is changed by waiting thread.
              */
             int ws = node.waitStatus;
             if (ws < 0)
                 compareAndSetWaitStatus(node, ws, 0);
        
             /*
              * Thread to unpark is held in successor, which is normally
              * just the next node.  But if cancelled or apparently null,
              * traverse backwards from tail to find the actual
              * non-cancelled successor.
              */
             Node s = node.next;
             if (s == null || s.waitStatus > 0) {
              
                 s = null;
                 for (Node t = tail; t != null && t != node; t = t.prev)
                     if (t.waitStatus <= 0)
                         s = t;
             }
             if (s != null)
                 LockSupport.unpark(s.thread);
         }

#### 1.3 AQS 的资源分配方式 ####

1.  **独占式 （Exclusive）**  
    对于独占式同步组件，同一时刻只有一个线程能获取到同步状态，其他线程都得去排队等待，其待重写的尝试获取同步状态的方法`tryAcquire()`返回值为 true 表示获取锁成功（这个操作会累加 state 处理锁重入的情况），同时会将获取到锁的线程设置为 `exclusiveOwnerThread`，只有 state 为 0 时才能获取到锁
2.  **共享式 （Share）**  
    对于共享式同步组件，同一时刻可以有多个线程同时获取到同步状态，通常将 state 初始化为某个整数，然后 CAS 操作 state 变量来实现线程间的协调

## 2. JDK 中 AQS 的应用 ##

#### 2.1 ReentrantLock ####

`ReentrantLock` 的 state 初始化为 0，表示未锁定状态。A线程lock()时，会调用tryAcquire()独占该锁并将state+1。此后，其他线程再tryAcquire()时就会失败，直到A线程unlock() 到state=0（即释放锁）为止，其它线程才有机会获取该锁。释放锁之前，A线程自己可以重复获取此锁（state累加），这是可重入的概念。但获取多少次就要释放多么次，这样才能保证state回到 0 值

#### 2.2 Semaphore ####

`Semaphore(计数信号量)` 允许多个线程同时访问，经常用于限制获取某种资源的线程数量。它将 state 初始化为某个整数作为共享资源，当有线程来获取锁时共享资源减少，释放锁则共享资源增加，共享资源不足时来获取锁的线程需要 park

#### 2.3 CountDownLatch ####

`CountDownLatch（倒计时器）`任务分为N个子线程去执行，state 也初始化为N（N要与线程个数一致）。这N个子线程并行执行，每个子线程执行完后countDown()一次，state就 CAS 减1。等到所有子线程都执行完后(即state=0)，会unpark()主调用线程，然后主调用线程就会从`await()`函数返回，继续运行

*  CountDownLatch是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当CountDownLatch使用完毕后，它不能再次被使用

#### 2.4 CyclicBarrier ####

`CyclicBarrier(循环屏障)` 和 CountDownLatch 非常类似，它也可以实现线程间的互相等待，但是它的功能比 CountDownLatch 更加复杂和强大。

CyclicBarrier 要做的事情是，让一组线程到达一个屏障（也可以叫同步点）之前被阻塞，直到最后一个线程到达屏障时，屏障才会打开，所有被屏障拦截的线程才能继续运行。CyclicBarrier 默认的构造方法是 CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用`await()`方法告诉 CyclicBarrier 自身已经到达了屏障，然后进入阻塞状态。

*  应用场景  
    CyclicBarrier 可以用于多线程计算数据，最后合并计算结果的应用场景。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/5c2b204d010546baa7f613943ae6e388.jpg