深入源码分析JUC框架中的AQS

矫情吗；* 2022-02-14 04:51 349阅读 0赞

# 深入源码分析J.U.C框架中的AQS #

### 文章目录 ###

*  深入源码分析J.U.C框架中的AQS
 *   *  1. 前言
     *   *  1.1 什么是AQS?
         *  1.2 为什么要学习AQS？
     *  2. 深入源码分析AQS
     *   *  2.1 AQS中的成员变量
         *  2.2 AQS中的内部类
         *  2.3 AQS中的方法
         *   *  2.3.1 state的线程安全
             *  2.3.2 AQS中比较常用且重要的方法
     *  3. 分析重入锁，理解AQS
     *   *  3.1 重入锁实战分析
     *  4. 总结
     *  5. ReentrantLock与Synchronized

## 1. 前言 ##

> 本篇文章需要你有使用过JUC同步组件，并了解什么是CAS。

### 1.1 什么是AQS? ###

AQS是一个JUC（java并发包）的一个类，全称是`AbstractQueuedSynchronizer`，它是同步组件（例如Lock、CountDownLatch）的一个基础框架，它是一个抽象类，使用了模版方法模式使继承于它的子类只需要实现一些方法即可简便的实现一个可靠的同步组件，由此特性可以看出，构建一个可靠的同步组件，你可以继承它并实现抽象方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70]  
由上图可以看出，大部分的同步组件都利用AQS去构建自己的基础同步框架，例如重入锁、读写锁、信号量、计数的`CountDownLatch`、`Condition`（类似Object的wait-notify），其中包括了公平锁与非公平锁，共享锁与非共享锁的实现，都由AQS内部组建起来，子类只需要维护状态的变化判断状态表现哪些行为即可，AQS可谓是一个强大的同步基础框架。

由实现来看，大部分同步组件都是在内部创建一个内部类去继承它，因为这样可以屏蔽内部细节，使使用者只需要关注同步组件的功能，不需要知道AQS的存在。

### 1.2 为什么要学习AQS？ ###

首先，作为一个强大的同步基础框架，其内部的实现大部分使用了CAS技术与volatile来保证内部的线程安全，可伸缩性很强，至少这点就值得我们去学习它。站在巨人的肩膀上，我认为有助于并发编程，写好一个线程安全类不仅仅只是让它线程安全，更要注意可伸缩性，简单理解就是并发的性能。

其次，它作为一个基础同步框架，JUC大部分同步组件都用到它的情况下，若能深刻理解它，变相的你也可以很容易理解JUC的同步组件的实现原理，有助于使用JUC同步组件。

最后，如果你想要编写一个自定义的同步组件，最好使用AQS来搭建你的基础（如果不使用AQS，做出一个可靠的同步组件是更难的），这时如果你不了解什么是AQS，是不可能做出一个可靠的同步组件的。

## 2. 深入源码分析AQS ##

> 下面提到的重入锁是ReentrantLock，以下的分析都会基于重入锁作示范，分析重入锁以理解AQS。

### 2.1 AQS中的成员变量 ###

首先来看几个AQS中比较重要的成员变量：

private transient volatile Node head;
    
    private transient volatile Node tail;
    
    private volatile int state;

*  Node head：头节点，AQS把线程封装成一个Node对象，Node对象是其内部类（下面会介绍），AQS内部维护一个队列的结构，使线程可以FIFO先进先出
 *  Node tail：尾节点，FIFO队列中最后的一个节点
 *  state：同步状态，在不同的同步组件中有不同的含义，例如重入锁中代表是否被独占，为1或大于1（重入）代表此时有锁，为0代表此时无锁。

### 2.2 AQS中的内部类 ###

static final class Node { 
    
    	//略过部分代码...
      
      //Node的前驱节点
      volatile Node prev;
    
      //Node的后继节点
      volatile Node next;
    
      //表示此Node代表的是哪个线程
      volatile Thread thread;
    
      //取出此Node的前驱节点
      final Node predecessor() throws NullPointerException { 
        Node p = prev;
        if (p == null)
          throw new NullPointerException();
        else
          return p;
      }
    
      Node() {     // Used to establish initial head or SHARED marker
      }
    
      Node(Thread thread, Node mode) {      // Used by addWaiter
        this.nextWaiter = mode;
        this.thread = thread;
      }
    
      Node(Thread thread, int waitStatus) {  // Used by Condition
        this.waitStatus = waitStatus;
        this.thread = thread;
      }
    }

其中，内部类`Node`维护这一个前驱节点，一个后继节点，看到这里，你或许已经发行了AQS中所维护的这个队列是一个链式结构：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70 1]  
大致是如上图一样的队列结构，在线程竞争时都是使用的这个队列结构来存储线程的顺序，控制线程获取锁。

### 2.3 AQS中的方法 ###

#### 2.3.1 state的线程安全 ####

AQS中有一个关键的成员变量`state`，其用来表示同步的状态，是一个会被经常更新的值，AQS是如何保证它的线程安全呢？

*  volatile关键字，确保其可见性，happens-before原则，确保一个线程的修改操作在一个线程读取之前，并且修改值对读取线程可见，这样确保了getState的可见性
    
        private volatile int state;
 *  CAS技术，确保其原子性，CAS是乐观锁的实现，如果更新不成功有冲突，就放弃此次更新，再重试，确保了修改时的原子性
    
        protected final boolean compareAndSetState(int expect, int update) { 
          // See below for intrinsics setup to support this
          // 底层使用unsafe工具类（JVM底层实现）完成CAS操作
          return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }

#### 2.3.2 AQS中比较常用且重要的方法 ####

*  将一个Node加入到队列的尾部：
    
        private Node enq(final Node node) { 
          // 无限循环
          // 在JVM底层中for语句块比while语句块少了几行机器指令，所以更快
          for (;;) { 
            // 先拿到此时的尾节点
            Node t = tail;
            // 如果尾部为空，代表AQS还是新的，先初始化
            if (t == null) {  // Must initialize
              // CAS将空Node设置为head
              if (compareAndSetHead(new Node()))
                // 如果成功，将空Node也设置成tail
                // 此时初始化之后，AQS的头尾都是空Node
                tail = head;
            } else { 
              // 将需要加入尾部的Node的前驱节点设为此时的tail
              node.prev = t;
              // CAS将tail设置为Node
              if (compareAndSetTail(t, node)) { 
                // 如果成功，tail的后继节点设置为node
                t.next = node;
                return t;
              }
            }
          }
        }
    
    假设此时只有一个线程在执行此方法，会将一个空Node设置为head，然后将自己这个Node设置为tail，在AQS中的队列结构如下图所示：
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70 2]  
    如果再有线程需要加入队列，则会加入到WaitAddNode之后，其变为tail，WaitAddNode为中间Node。这一过程是线程安全的，因为设置尾节点操作是CAS操作，如果有一个线程正在竞争设置tail，此时有一个线程会失败，然后for循环再次尝试CAS设置tail，此时失败的线程是可以看到新tail并在其后添加。只有设置了tail，线程才会从循环中退出。
 *  将当前线程包装为Node，添加进至AQS队列尾部：
    
        private Node addWaiter(Node mode) { 
          // 将当前线程包装成Node对象
          Node node = new Node(Thread.currentThread(), mode);
          // Try the fast path of enq; backup to full enq on failure
          // 这里先尝试插入队列尾部，如果失败了反正也有上面enq方法继续自旋插入
          // 与上面的enq方法类似，这里只尝试一次，后续在enq继续尝试
          Node pred = tail;
          if (pred != null) { 
            node.prev = pred;
            if (compareAndSetTail(pred, node)) { 
              pred.next = node;
              return node;
            }
          }
          enq(node);
          return node;
        }
    
    这里只比enq多了一步包装线程为Node对象，不作过多分析。
 *  尝试获取锁：
    
    在AQS中，Head的Node节点通常是获得资格正在运行的线程（重入锁中）。线程释放需要两个条件，1.当前节点的前驱节点为head，2.子类需要实现的方法tryAcquire返回true，才可以获取到锁，线程才能继续执行
    
        final boolean acquireQueued(final Node node, int arg) { 
          boolean failed = true;
          try { 
            boolean interrupted = false;
            // 同样的自旋操作
            for (;;) { 
              // 调用了之前提到的方法，获取node的前驱节点
              final Node p = node.predecessor();
              // 这里是判断此Node是否有资格被释放
              // 1.前驱节点为head
              // 2.tryAcquire方法（子类实现）返回true
              if (p == head && tryAcquire(arg)) { 
                // 到这里说明有资格获取锁了，将Node设置为head
                setHead(node);
                // 此时p为node的前驱节点
                // node获取到锁之后说明p这个node已经没有用了，释放它的next指针
                // 方面后续GC清理p
                p.next = null; // help GC
                failed = false;
                return interrupted;
              }
              // 自旋次数过多，此时这里会park阻塞线程，让线程先停一停
              // 如果不断自旋，将一直占用CPU，这是不好的
              // 阻塞线程之后，将Node中表示的状态变为SIGNAL
              // 后续会对SIGNAL的Node进行unpark释放线程
              if (shouldParkAfterFailedAcquire(p, node) &&
                  parkAndCheckInterrupt())
                interrupted = true;
            }
          } finally { 
            if (failed)
              cancelAcquire(node);
          }
        }
    
    这里基本就是将Node设置为AQS中的head节点而已，在后续head节点是被执行的节点，所以变相的这里就是获取锁的过程。
 *  子类需要实现的几个方法：
    
     *  tryAcquire：以重入锁（公平锁）为例
    
        protected final boolean tryAcquire(int acquires) { 
          final Thread current = Thread.currentThread();
          // 获取当前的state
          int c = getState();
          // 如果为0，表示此时锁没有人占用
          if (c == 0) { 
            // CAS将1设置到state中，若成功表示成功占用
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) { 
              // 将此时占用的线程设置为当前线程
              // 表示当前线程占用了此锁
              setExclusiveOwnerThread(current);
              return true;
            }
          }
          // 如果到了这里，说明锁已经被人占用了
          // 此时看看占用锁的线程是否是当前线程
          // 如果是当前线程，state自增1，这里就是重入的过程
          // 如果是重入的话，也表示可以获取到锁，返回true
          else if (current == getExclusiveOwnerThread()) { 
            // state加一
            int nextc = c + acquires;
            if (nextc < 0)
              throw new Error("Maximum lock count exceeded");
            // 将新state设置到AQS中
            // 此时为什么不用CAS，因为如果程序能走到这里，说明此时只有一个线程
            // 不存在线程安全问题，直接设置，优化速度
            setState(nextc);
            return true;
          }
          // 获取失败
          return false;
        }
        }
    
    以上是ReentrantLock重入锁中的实现，可以看出其tryAcquire方法对state的判断即为0表示无锁，可以获取到，为1以上表示有锁，拒绝获取。当然不同同步组件此方法的实现是不同的，若你想要实现一个双锁占用（有两把锁，只有最多两个线程才能执行），此时你就可以判断state<2拒绝获取。
    
     *  tryRelease：还是以重入锁为例
        
            protected final boolean tryRelease(int releases) { 
              // 获取state，减去释放数（重入锁一般为1）
              int c = getState() - releases;
              // 必须是获取了锁的线程才能进行释放操作，不然抛异常
              if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
              boolean free = false;
              // 若此时state减1为0的话，代表锁需要被释放了
              if (c == 0) { 
                free = true;
                // 将线程占用的变量设为null，代表现在没有线程占用
                setExclusiveOwnerThread(null);
              }
              // 设置新state
              // 只有获取了锁的线程才可以设置，所以此时单线程操作
              // 无需考虑线程安全问题
              setState(c);
              return free;
            }
        
        一般来说，是将state减1，在独占锁中还会把线程占用标志设为null
     *  上面两个方法是独占锁的获取与释放，当然还有共享锁的获取与释放，这里就不一一解释了，大致都是对state进行判断然后操作，不同的state判断体现了锁的不一样的行为（共享或独占）
 *  整合以上子方法，构造一个真正获取锁的入口方法：
    
        public final void acquire(int arg) { 
          if (!tryAcquire(arg) &&
              acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
        }
    
    首先调用子类需要实现的也就是上面提到的方法tryAcquire，先尝试获取锁，如果获取得到，不会执行后面的方法。若获取不到，将现执行addWaiter方法将当先线程包装为Node，然后加入AQS的队列尾部，acquireQueued、addWaiter这两个方法上面都有介绍。
 *  释放锁，在重入锁中就是unlock：
    
        public final boolean release(int arg) { 
          // 此方法上面提到，是子类需要实现的方法
          if (tryRelease(arg)) { 
            // 若释放锁成功，则unpark后续节点（如果被park了的话）
            Node h = head;
            if (h != null && h.waitStatus != 0)
              // 唤醒后续节点，使后续节点可以醒来争夺锁
              unparkSuccessor(h);
            return true;
          }
          return false;
        }
    
    可以看到，主要释放逻辑还是在子类自己实现，AQS只是大致将head的后续节点唤醒，使其醒来争夺锁（前面提到自旋过多会park阻塞，在此时会unpark唤醒，继续自旋）

## 3. 分析重入锁，理解AQS ##

到这里分析了AQS中一些主要的方法，这里结合重入锁的实现来看，可以更好的理解AQS。

> 重入锁中，还有公平与不公平的区别，具体差异是公平锁会严格按照AQS的队列进行获取锁，只有Head节点的线程才可以获取的到锁，而非公平锁的话如果当前线程获取时正好有锁释放，他就能“插队”获取锁，所以叫做非公平锁，在实际环境中，非公平锁的性能在大部分场景下是优于公平锁的，所以非公平锁也是重入锁中的默认实现。

在重入锁的实现中，其内部类继承了AQS，实现了上述几个需要子类实现的方法：

abstract static class Sync extends AbstractQueuedSynchronizer { 
    
      // 略过部分方法...
      
      // 这里tryAcquire方法在内部类的子类中实现
      // 因为公平锁与非公平锁获取锁的方式是不一样的
      // 所以具体是在具体子类中不同实现
      
      // 释放锁方法
      protected final boolean tryRelease(int releases) { 
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
          throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) { 
          free = true;
          setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
      }
    }

### 3.1 重入锁实战分析 ###

先来看一段使用重入锁的场景：

public class ReentrantLockTest { 
        
        private static Lock lock;
        
        public static void main(String[] args) { 
            
            lock = new ReentrantLock();
            
            try { 
                lock.lock();
                
                // xx操作
            }finally { 
                lock.unlock();
            }
        }
    }

这里调用了ReentrantLock的lock方法，让我们进去看看发生了什么：

public void lock() { 
      sync.lock();
    }

调用了我们刚刚提到的内部类的lock方法，继续进去看看（这里以非公平锁为示例，毕竟是默认实现嘛）：

final void lock() { 
      // 直接CAS操作，如果state为0则将state设置为1
      if (compareAndSetState(0, 1))
        // 如果成功设置，当前线程直接占用锁（插队）
        setExclusiveOwnerThread(Thread.currentThread());
      else
        // 如果CAS失败，则乖乖排队
        acquire(1);
    }

这里可以看出，在非公平锁中获取锁的插队行为，在公平锁中获取锁行为是乖乖按照AQS的步骤走的：

public final void acquire(int arg) { 
      if (!tryAcquire(arg) &&
          acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
    }

以上这个方法就是我们前面分析的AQS的方法acquire，上面分析过了，所以这里就简单带过一下：

1.  首先先获取state，为0则可以获取锁，为1则获取锁失败
2.  若获取锁失败，则后续addWaiter方法将当前线程封装为Node并加入AQS队列的尾部，acquireQueued方法将自旋获取锁，自旋一会会阻塞线程让出CPU。

只有获取了锁的线程，acquire方法才能继续执行下去，才会到我们上面demo写的xx操作中，重入锁的lock方法实现了只有一个线程才可以在同步块执行，实现了与关键字synchronized一样的效果。

以上是同步锁线程，那么下面是锁释放。

回到demo中的finally块中，调用了lock.unlock方法，进去看看：

public void unlock() { 
      sync.release(1);
    }

这里依然是调用了内部类的方法：

public final boolean release(int arg) { 
      if (tryRelease(arg)) { 
        Node h = head;
        if (h != null && h.waitStatus != 0)
          unparkSuccessor(h);
        return true;
      }
      return false;
    }

同样，在上面我们也分析过此方法，是AQS中的模版方法，这里简略分析带过。

首先调用tryRelease方法，此方法在子类中得以实现：

protected final boolean tryRelease(int releases) { 
      int c = getState() - releases;
      if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
      boolean free = false;
      if (c == 0) { 
        free = true;
        setExclusiveOwnerThread(null);
      }
      setState(c);
      return free;
    }

这段代码也讲过了，主要是判断state若为0，将线程占用标志清除（设为null），若state为0，则表示可以释放锁，free标志为true，返回true。

回到上面的release方法，若tryRelease返回true则代表此时可以释放锁，此时判断head的后续节点是否被park了，如果被park了就unpark它，使其醒来争夺锁。

此时会发生什么呢？首先tryRelease中将独占标志设置为null，且state为0，然后唤醒head的后续节点，此时这段阻塞代码将会唤醒运行：

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);
      }
    }

此时获取锁资格之一：前驱节点为head条件已经满足，则执行tryAcquire方法看看是否有获取锁资格之二：

protected final boolean tryAcquire(int acquires) { 
      final Thread current = Thread.currentThread();
      int c = getState();
      // 此时state为0
      if (c == 0) { 
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) { 
          // 成功将当前线程设置为独占标志
          setExclusiveOwnerThread(current);
          return true;
        }
      }
      else if (current == getExclusiveOwnerThread()) { 
        int nextc = c + acquires;
        if (nextc < 0)
          throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
      }
      return false;
    }

看到这里的读者应该明白了，比如现在有A B两个线程正在执行demo方法，A线程先获取到锁，之后B线程尝试获取，B线程自旋获取锁然后阻塞，A线程调用unlock释放锁，B线程则调用上面的方法，成功获取到锁，B线程将继续执行下去，实现了锁的功能。

## 4. 总结 ##

到这里，我们分析了AQS中的几个成员变量，几个内部类和几个重要的方法，并且结合重入锁ReentrantLock的实现来分析AQS在同步组件中的作用：

AQS提供了一个队列来存放排队的线程，提供了一些方法使线程可以在队列中阻塞，释放锁之后可以使阻塞的线程开始执行，其中还有很多功能全面的方法没有分析到，例如定时获取锁，其实就是在自旋过程中判断获取的时间，每自旋一次判断一次，若超时则会返回。或是共享锁的获取、读写锁的实现，有兴趣的读者可以自行阅读源码，相信懂得了以上AQS分析之后，再去看别的同步组件的实现并不是一件难事。

## 5. ReentrantLock与Synchronized ##

这里引申一个题外问题，ReentrantLock与Synchronized的区别与如何选择？

我们分析了ReentrantLock显式锁的底层实现，发现其功能其实与Synchronized差不多，那么这两者有什么区别呢？

1.  在性能上其实两者是差不多的（Synchronized在后续有做性能优化，加入了锁策略提升了性能），所以在性能上没什么区别
2.  在功能上，ReentrantLock比Synchronized多了很多功能，例如定时获取锁、公平、非公平的选择、获取多个等待队列（Condition）、可中断的获取锁、可判断锁的状态

ReentrantLock功能如此强大，是否直接选择其就好了？

其实也不是，各有各的优缺点：

*  ReentrantLock
    
     *  优点：功能多、灵活
     *  缺点：需要显式调用unlock解锁，不然是一颗定时炸弹
 *  Synchronized
    
     *  优点：简单、易于编写，自动释放锁解决隐患。可读性强，有些程序员只用过Synchronized，并不认识JUC的同步工具
     *  缺点：功能不全，不能灵活中断、定时之类的，wait-notify机制在多条件队列中显得很不好用

鉴于优缺点的分析，我们可以得到以下结论：

应该优先使用Synchronized，当出现以下需求时应该使用ReentrantLock

*  需要额外的定时、可中断之类的功能
 *  需要使用wait-notify的功能，最好使用Condition，所以需要使用ReentrantLock作为锁的实现

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70]: /images/20220214/03b4a0c9ccb343a2b71e5ab791580a95.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70 1]: /images/20220214/07fbecaaeeaf4a95a476314d3b281e2d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70 2]: /images/20220214/a26bf8b40cd74e24b66044fc1951c2a5.png