AbstractQueuedSynchronizer即AQS队列同步器，可能有不少朋友听说过这个名词，但是不了解它的用途、原理，也可能听都没有听说过这个类…(面壁吧…) 但要提到ReentrantLock，那大家应该就不会陌生了。
没错！ReentrantLock就是基于AQS实现的，不仅仅是ReentrantLock，还有诸如Lock、Semaphore、CountDownLatch的底层都是靠它实现的，也许平时在使用的时候，直接一个lock()锁住，然后执行完业务代码后，一个unlock()解锁，相当方便！但优秀的你绝对不会错过了解它们的底层实现原理的机会吧！
很好，如果有兴趣就接着往下看呗~

既然大家更加熟悉ReentrantLock，那么以ReentrantLock为切入点来讲解AQS再好不过了~
ReentrantLock被称之为可重入锁，属于悲观锁的一种(关于什么是乐观锁，什么是悲观锁这里就不细讲啦)。本文主要会从线程获取锁的流程，获取锁失败后的入队、出队流程，队列结构，以及锁的释放等方面做详细的分析。

ReentrantLock通常用来和Synchronized作比较，其实在性能方面，Synchronized和ReentrantLock基本上是持平的，但是相比Synchronized，后者的功能更加的丰富，操作更加灵活，因此比较适合复杂的并发场景。

通过下面的代码，大家可以对比下

// ---------------------Synchronized------------------------------------
public static void testSynchronized() {
   for (int i = 0; i < 100; i++) {
       // 可重入，可用于方法或者代码块
       synchronized (LockTest.class) {
           LOGGER.info("get syn lock[{}]",i);
       }
   }
}
// ---------------------ReentrantLock-----------------------------------
public static void testReentrantLock() throws InterruptedException{
   // 初始化可重入锁并选择锁类型 true：公平锁 false：非公平锁
   ReentrantLock lock = new ReentrantLock(true);
   for (int i = 0; i < 100; i++) {
       // 手动上锁
       lock.lock();
       try {
           try {
               // 尝试获取锁，等待时间1000毫秒(可支持多种加锁方式，比较灵活; 具有可重入特性)
               if (lock.tryLock(1000, TimeUnit.MILLISECONDS)) {
                   LOGGER.info("get reentrant lock[{}]",i);
               }
           } finally {
               // 手动释放锁
               lock.unlock();
           }
       } finally {
           lock.unlock();
       }
   }
}

很明显可以看出，当使用ReentrantLock时，在lock()和unlock()之间，可以穿插tryLock()等方法，根据业务需求灵活多变的处理。以上代码执行完之后的结果如下，可以看到性能方面都差不多的…

AQS，AbstractQueuedSynchronizer，即队列同步器。它是构建锁或者其他同步组件的基础框架（如ReentrantLock、ReentrantReadWriteLock、Semaphore等），JUC并发包的作者（Doug Lea）期望它能够成为实现大部分同步需求的基础。它是JUC并发包中的核心基础组件。

ReentrantLock相比Synchronized，除了支持非公平锁，还支持公平锁，公平锁是指当锁可用时,在该锁上等待时间最长的线程将获得锁的使用权(先来后到)。而非公平锁则随机分配这种使用权。很显然，使用随机的方案带来的好处是性能更高。以下是创建公平锁和非公平锁的方式。

// 创建公平锁
ReentrantLock lock = new ReentrantLock(true);
// 创建非公平锁
ReentrantLock lock = new ReentrantLock(false);

具体的构造函数源码如下，不难看出fair的值会控制ReentrantLock的类型

public ReentrantLock(boolean fair) {
	// 通过fair值，来创建不同的锁对象
    sync = fair ? new FairSync() : new NonfairSync();
}

我们可以做个小实验，看看它们两者的区别。先来使用公平锁测试，同时创建5个线程去获取锁，并且给每个线程两次获取锁的机会，注意观察线程获取锁的规则。

public class ReentrantLockTest {
	// 创建公平锁
    static Lock lock = new ReentrantLock(true);

    public static void main(String[] args) {
		// 创建五个线程执行任务
        for (int i = 0; i < 5; i++) {
            new Thread(new ThreadDemo(i)).start();
        }
    }

    static class ThreadDemo implements Runnable {
        private static final Logger LOGGER = LoggerFactory.getLogger(ThreadDemo.class);
        private Integer id;
        
        public ThreadDemo(Integer id) {
            this.id = id;
        }

        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            for (int i = 0; i < 2; i++) {
                lock.lock();
                LOGGER.info("线程[{}]获得锁", id);
                lock.unlock();
            }
        }
    }
}

通过运行结果不难看出，这5个线程每个线程都获取释放锁两次，并且是轮流有序的获取到锁。我们再将参数fair改为false测试一下非公平锁，看看效果。

此时的线程会重复获取锁。假设此时有很多个线程去获取锁，并且同一个线程获取锁的机会更多，那么可能会造成某些线程长时间得不到锁，就好比你去KFC点餐，可能你前面的那个人随同的伙伴有很多，本来他已经把他自己的餐点好了，结果他的小伙伴说，伙计帮我也点一份，虽然他刚买完单，但是位置还是排在第一个，那么他就可以继续点了，当然后面还在排队的人肯定会不满，毕竟这跟插队没两样… 还可能导致纠纷…
这就是非公平锁的“饥饿”问题所在。

说了这么多ReentrantLock的公平锁和非公平锁，主要是为了研究AQS原理做个铺垫，通过FairSync和NonfairSync的实现，我们可以更深入的了解ReentrantLock和AQS的关系，以及AQS的底层原理。其实在刚刚的实验中，我们已经发现了公平锁和非公平锁的不同之处。既然ReentrantLock的底层是由AQS来实现的，那么公平锁和非公平锁是如何通过AQS实现的呢？我们不得不打开ReentrantLock的源码一探究竟啦！

非公平锁源码中的加锁流程如下

final void lock() {
  	// 若通过CAS设置变量State，成功则设置线程为独占锁
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
    	// 失败则调用acquire方法，进入后续的锁获取候选流程
        acquire(1);
}

通过CAS(Compare And Swap)方式设置变量State(同步状态)，若成功则设置当前线程为独占锁，若失败，则调用acquire方法，进入后续的锁获取候选流程。获取独占锁，顾名思义就是锁被这个线程独占了，其他的线程得等到我释放锁之后才能去获取，这个很好理解。那么获取失败后的线程将何去何从呢？
其实我们试着猜想一下，也能得出结论。无非就两种情形，要么获取锁失败之后，将状态位设置为失败，或者放弃，并结束锁获取流程。但是这种设计显然不符合我们的设计理念，极大的降低系统并发，并极有可能造成大量的请求无法正常处理并返回结果。
显然，为了解决这样的问题，就一定会存在某种队列结构，来接收获取锁失败的线程。如果真的是如我们所猜想有这么个结构存在，那么在队列中的线程何时可以在此获取到锁呢？如果再次获取失败该如何处理呢？假设一直失败一直重入队列等待，会不会造成死循环问题？
这所有谜团的答案都在acquire(1)方法中，那么带着这些疑问我们再来看公平锁源码。

公平锁源码中的加锁流程如下

发现公平锁中的lock()加锁，并没有CAS判断，而是直接调用acquire(1)方法？
好啦，就不卖关子了，我们直接分析acquire(1)的源码。

acquire(1)中会先尝试获取独占锁，调用tryAcquire方法，如果获取成功则直接返回，如果未获取成功，则将当前线程包装成Node，并加入同步队列等待(在队列中会检测是否前驱为HEAD，并尝试获取锁，如果获取失败，则会通过LockSupport阻塞当前线程，直至被释放锁的线程唤醒或者被中断，随后再次尝试获取锁，并反复执行该流程）其实通过这个方法就更加印证了我们之前关于等候队列的猜想是正确的。selfInterrupt()意为产生一个中断，如果在acquireQueued()中当前线程被中断过，则需要产生一个中断(画个圈，后面会重点说明为什么这个地方需要手动去中断线程)。为了更好的理解源码中的方法，我会通过绘图的方式来加深大家的理解。

通过上面的分析，我们已经知道获取锁失败的线程会加入到队列中，这样做的好处是避免锁被释放的瞬间，所有的线程都去争抢资源，减少并发冲突，避免了"惊群效应"，假设有10000个线程等待获取锁，当锁被释放后，只会通知队列中的第一个线程去竞争锁。那么这个队列到底是个什么结构呢？

CLH：Craig、Landin and Hagersten队列，是单向链表，AQS中的队列是CLH变体的虚拟双向队列（FIFO），当多线程争用资源被阻塞时会进入此队列，AQS是通过将每条请求共享资源的线程封装成一个节点来实现锁的分配。

为什么说AQS中的队列是CLH变体的虚拟双向队列呢？因为AQS中的队列不存在队列实例，仅存在结点之间的关联关系，原始CLH队列，一般用于实现自旋锁，而在变种CLH队列中，获取不到锁的线程，一般会时而阻塞，时而唤醒。AQS还维护了一个volatile修饰的State同步状态位(代表共享资源)，对每一个线程都是可见的，并通过内置的FIFO队列来完成资源获取的排队工作，通过CAS完成对State值的修改(CAS操作必须得依靠volatile修饰的变量来实现同步)。
还有一个需要关注的点就是队列中的Node(即CLH变体队列中的节点)，它实质上是对线程的一个封装。

当然，细心的各位也应该发现了addWaiter()方法中传入的Node为Node.EXCLUSIVE，它表示线程正在以独占的方式等待锁，当然除了独占锁，还有一种模式为Node.SHARED，表示线程以共享的模式等待锁。线程在队列的等待状态waitStatus在AQS中也是非常重要的概念，直接决定了哪些线程可以获取锁，哪些线程需要继续等待，哪些线程要被cancel并淘汰。

刚刚我们提到的AQS维护的一个volatile修饰的State同步状态位，主要用于展示当前资源的获取锁的情况，独占模式下和共享模式下，根据State同步状态位判断获锁情况的方式不一样，独占模式下，会初始化状态位State为0，此时当线程尝试去获取锁时，会判断当前的State的值是否为0，如果为0说明锁没有被获取，那么该线程获取锁之后设置State为1，此时当其他线程来获取锁时，发现状态位为1，就去排队了。
而共享模式下，代表可以有多个线程一起共享锁，但是很显然这个可获取锁的线程数量是有限制的，那么此时获取锁的规则是，先初始化State的值，每当一个线程获取到了共享锁之后，就会减少State的值，当State小于等于0之后，且其他拥有锁的线程没有释放锁的情况下，当前线程就会被阻塞。

在AQS中提供下面几个方法用于获取及操作状态位State。

说了这么多关于AQS的结构，也是为了下面深入AQS源码讲解做铺垫。我们继续回到ReentrantLock的acquire(1)方法上来。刚才我们讲到当acquire(1)方法中的tryAcquire()方法返回了True，则说明当前线程获取锁成功，如果获取失败，就需要加入到等待队列中。在非公平锁中的tryAcquire()本质上是调用nonfairTryAcquire(int acquires)，acquires的值为1，关于该方法的分析如下。

final boolean nonfairTryAcquire(int acquires) {
     final Thread current = Thread.currentThread();
     // 获取状态位
     int c = getState();
     if (c == 0) {
     	// 当状态位为0，通过CAS方式判断当前线程是否被篡改，若没有则设置当前线程独占锁
         if (compareAndSetState(0, acquires)) {
             setExclusiveOwnerThread(current);
             return true;
         }
     }
     // 若当前线程已经持有锁了
     else if (current == getExclusiveOwnerThread()) {
     	// 则设置状态位为1
         int nextc = c + acquires;
         if (nextc < 0) // overflow
             throw new Error("Maximum lock count exceeded");
         setState(nextc);
         return true;
     }
     // 以上条件都不满足则返回false，阻塞线程
     return false;
 }

tryAcquire(int arg)方法在AQS源码中只是简单的抛了一个异常，因此上面的tryAcquire(int arg)方法其实已经被ReentrantLock重写了。

而在公平锁中的tryAcquire(int arg)方法就改动了下面这部分，添加了hasQueuedPredecessors()判断。用于加锁时判断等待队列中是否存在有效节点的方法。如果返回false，说明当前线程有去争取锁的资格，如果返回true，说明队列中存在有效节点，当前线程必须加入到等待队列中。

final boolean nonfairTryAcquire(int acquires) {
    ...
    if (c == 0) {
    	// 当状态位为0，通过CAS方式判断当前线程是否被篡改，若没有则设置当前线程独占锁
       if (!hasQueuedPredecessors() &&
          compareAndSetState(0, acquires)) {
           setExclusiveOwnerThread(current);
           return true;
       }
    }
    ...
}

通过上面的分析，我们已经知道了线程是如何通过修改状态位的值判断是否能获取锁的，以及获取锁失败之后会进入等待队列中，同时也大致清楚队列的结构了，那么将线程添加到队列这个动作具体在源码中怎么体现的呢？
我们先来研究acquireQueued(addWaiter(Node.EXCLUSIVE), arg)中的addWaiter(Node mode)方法。

// 将获取锁失败的线程置入队列中
private Node addWaiter(Node mode) {
	// 将当前的线程封装为NODE节点，模式为 EXCLUSIVE 独占模式
    Node node = new Node(Thread.currentThread(), mode);
    // 获取尾节点
    Node pred = tail;
    if (pred != null) {
    	// 如果尾节点不为空，则设置当前节点的前一个节点为之前获取的尾节点
        node.prev = pred;
        // 通过CAS的方式设置之前尾节点的下一个节点为当前节点
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 如果尾节点为null，执行该方法
    enq(node);
    return node;
}
// 通过CAS的方式设置尾节点，对tailOffset和Expect进行比较
// 如果tailOffset的Node和Expect的Node地址是相同的，那么设置Tail的值为Update的值
private final boolean compareAndSetTail(Node expect, Node update) {
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}
// 当尾节点为null时，通过循环+CAS在队列中成功插入一个节点后返回
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        // 初始化head和tail，这个条件是必须的！
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
        	// 通过CAS的方式设置之前尾节点的下一个节点为当前节点
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

通过上方的源码很明显可以看出，这个队列是一个双向队列，并且新增节点都是往尾部进行添加的。可能大家会有疑惑，就是addWaiter和enq方法中新增一个节点时为什么要先将新节点的prev设为tail再尝试CAS，而不是CAS成功后再进行双向链接操作？(如下所示) 这样会不会更安全？

if (compareAndSetTail(pred, node)) {
 	node.prev = pred;
    pred.next = node;
    return node;
}

因为双向链表目前没有基于CAS原子插入的手段，如果按照上面的代码去执行，此时pred为方法执行时读到的tail，引用封闭在栈上，会导致这一瞬间的tail也就是pred的prev为null，这就使得这一瞬间队列处于一种不一致的中间状态。因此，AQS通过将这两个操作(node.prev = pred和pred.next = node)分离在CAS前后的好处是，保证每时每刻的tail.prev都不会是一个null值(保证队列的完整)。其实细品AQS的源码会发现很多套路。

在enq(final Node node)这个方法中还有一点需要注意的是，当获取不到tail时，会初始化一个Node为head头节点，并且使tail指向head，但此时的头结点并不是当前线程节点，而是调用了无参构造函数的节点，也就是该节点不持有线程，不存储任何信息，只是占位。因此，真正的第一个存储数据的节点，是在第二个节点开始的

综上我们可以完善一下之前画的原理图。

上文解释了addWaiter方法，该方法就是将没有获取到锁的线程以Node的数据结构加入一个双向队列中，并返回一个包含线程的Node节点，再回到acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法，该方法可以让排队中的线程不断的进行"获锁"操作，直到获取锁(中断)。

final boolean acquireQueued(final Node node, int arg) {
    // 标记线程是否获锁成功
    boolean failed = true;
    try {
        // 标记线程在等待过程中是否中断过
        boolean interrupted = false;
        // 开始自旋，要么获取锁，要么中断
        for (;;) {
            // 获取当前节点的前驱节点
            final Node p = node.predecessor();
            // 如果p是头结点(虚节点)，说明当前节点在真实数据队列的首部，就尝试获取锁
            if (p == head && tryAcquire(arg)) {
                // 获取锁成功，头指针移动到当前node
                setHead(node);
                // 通知GC回收
                p.next = null; 
                failed = false;
                return interrupted;
            }
            // 执行到此，说明p为头节点且当前没有获取到锁（可能是非公平锁被抢占了）或者是p不为头结点，这个时候就要判断当前node是否要被阻塞（被阻塞条件：前驱节点的waitStatus为-1），防止无限循环浪费资源。
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

以上源码中的setHead(node)操作是将当前已经获取到锁资源的节点设置为虚节点(清空数据，并将head的指针指向该节点)，但是并没有对该节点的waitStatus进行修改，原因是这个节点还需要继续被使用到(除非已经不使用了，比方说它的下一个节点变成了头节点)。

当p为头节点且当前没有获取到锁（可能是非公平锁被抢占了）或者是p不为头结点时会执行下面的方法，判断当前线程是否应该被阻塞。

// 靠前驱节点判断当前线程是否应该被阻塞
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取头结点的节点状态
    int ws = pred.waitStatus;
    // 当头结点处于唤醒状态时，需要阻塞线程，防止无限循环浪费资源
    if (ws == Node.SIGNAL)
        return true; 
    // 通过枚举值我们知道waitStatus>0即waitStatus=1是取消状态
    if (ws > 0) {
        do {
            // 循环向前查找取消节点，把取消节点从队列中剔除
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 设置前任节点等待状态为SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

parkAndCheckInterrupt方法主要用于挂起当前线程，阻塞调用栈，并返回当前线程的中断状态。

private final boolean parkAndCheckInterrupt() {
	// 挂起当前线程
    LockSupport.park(this);
    return Thread.interrupted();
}

因此，我们可以得出结论，结束循环并出队的条件是，前驱节点为头节点，并且当前的线程成功获取到了锁，并且我们可以通过判断前驱节点的状态(是否是被唤醒状态)来决定是否挂起当前线程，而不用进行循环获取前驱状态的操作，适当的减少了资源的消耗。总结一下上面的操作如下图所示。

通过上面的分析，我们发现当获取前驱节点的状态时，会有一部分节点的状态是1，表示该节点取消循环获取操作，那么CANCELLED状态节点是如何生成的呢？带着这个疑问，我们可以观察acquireQueued()中的cancelAcquire(node)方法，这个方法在finally代码中。

// 取消继续获取前驱节点状态操作
private void cancelAcquire(Node node) {
    if (node == null)
        return;
  	// 设置该节点为虚节点
    node.thread = null;
    Node pred = node.prev;
  	// 通过判断前驱节点的状态，跳过取消状态的node
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;
  	// 获取过滤后的前驱节点的后继节点
    Node predNext = pred.next;
  	// 把当前node的状态设置为CANCELLED
    node.waitStatus = Node.CANCELLED;
  	// 如果当前节点是尾节点，将从后往前的第一个非取消状态的节点设置为尾节点
  	// 更新失败的话，则进入else，如果更新成功，将tail的后继节点设置为null
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        int ws;
    	// 如果当前节点不是head的后继节点
    	// 则判断当前节点前驱节点的是否为SIGNAL，如果不是，则把前驱节点设置为SINGAL看是否成功
    	// 如果上述操作中有一个为true，再判断当前节点的线程是否不为空
    	// 如果上述条件都满足，把当前节点的前驱节点的后继指针指向当前节点的后继节点
        if (pred != head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
      		// 如果当前节点是head的后继节点，或者上述条件不满足，那就唤醒当前节点的后继节点
      		// 主要是为了保证队列的活跃性，需要去唤醒一次后继线程，比如说当pred == head时
      		// 可能此时并没有线程持有锁，更不会去唤醒后继线程，那么如果没有这个操作，就会导致后面的线程
      		// 都不会被唤醒，整个队列就会挂掉了
            unparkSuccessor(node);
        }
        node.next = node; // help GC
    }
}

上面的流程比较复杂，我们再来梳理一下，首先获取当前节点的前驱节点，如果前驱节点的状态是CANCELLED，就一直往前遍历，直至找到第一个状态不为CANCELLED的节点，并将当前节点和找到的节点相关联，并将当前的节点的状态设置为CANCELLED。
以上操作完成之后，根据当前的节点的位置可以分为三种情况

• 节点为尾节点
• 节点是头节点的后继节点
• 节点既不是头节点的后继节点，也不是尾节点
  

在不同的情形下，都是对节点的next指针进行操作，但是并没有对prev指针进行操作？

1）执行cancelAcquire的时候，当前节点的前置节点可能已经从队列中出去了（已经执行过Try代码块中的shouldParkAfterFailedAcquire方法了），如果此时修改Prev指针，有可能会导致Prev指向另一个已经移除队列的Node，因此这块变化Prev指针不安全。

2）shouldParkAfterFailedAcquire方法中，会执行下面的代码，其实就是在处理Prev指针。shouldParkAfterFailedAcquire是获取锁失败的情况下才会执行，进入该方法后，说明共享资源已被获取，当前节点之前的节点都不会出现变化，因此这个时候变更Prev指针比较安全。

                            
                            版权声明：本文来源CSDN，感谢博主原创文章，遵循 CC 4.0 by-sa 版权协议，转载请附上原文出处链接和本声明。

                            原文链接：https://blog.csdn.net/weixin_44698119/article/details/103440871

                            站方申明：本站部分内容来自社区用户分享，若涉及侵权，请联系站方删除。