AQS的实现原理

原文：https://www.cnblogs.com/sglx/p/15190246.html

一、简介

　　AQS全称为AbstractQueuedSynchronizer，它提供了一个FIFO（First in First out 先入先出）队列，可以看成是一个用来实现同步锁以及其他涉及到同步功能的核心组件，常见的有:ReentrantLock、CountDownLatch等。AQS是一个抽象类，主要是通过继承的方式来使用，它本身没有实现任何的同步接口，仅仅是定义了同步状态的获取以及释放的方法来提供自定义的同步组件。可以这么说，只要搞懂了AQS，那么J.U.C中绝大部分的api都能轻松掌握。

从使用层面来说，AQS的功能分为两种：独占和共享

独占锁：每次只能有一个线程持有锁，比如ReentrantLock就是以独占方式实现的互斥锁。

共享锁：允许多个线程同时获取锁，并发访问共享资源，比如ReentrantReadWriteLock。

二、AQS的基本属性和方法

//头节点
private transient volatile Node head;
//尾节点
private transient volatile Node tail;
//状态值
private volatile int state;

　　AQS的实现依赖内部的同步队列，也就是FIFO的双向队列，如果当前线程竞争锁失败，那么AQS会把当前线程以及等待状态信息构造成一个Node加入到同步队列中，同时再阻塞该线程。当获取锁的线程释放锁以后，会从队列中唤醒一个阻塞的节点(线程)。

　　AQS队列内部维护的是一个FIFO的双向链表，这种结构的特点是每个数据结构都有两个指针，分别指向直接的后继节点和直接前驱节点，所以双向链表可以从任意一个节点开始很方便的访问前驱和后继。每个Node其实都是由线程封装的，当线程争抢锁失败后会封装成Node加入到ASQ队列中去。

Node类的组成如下

static final class Node {
        static final Node SHARED = new Node();
        static final Node EXCLUSIVE = null;
        static final int CANCELLED =  1;
        static final int SIGNAL    = -1;
        static final int CONDITION = -2;
        static final int PROPAGATE = -3;
        volatile int waitStatus;
        volatile Node prev; //前驱节点
        volatile Node next; //后继节点
        volatile Thread thread;//当前线程
        Node nextWaiter; //存储在condition队列中的后继节点
        //是否为共享锁
        final boolean isShared() { 
            return nextWaiter == SHARED;
        }

        final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }

        Node() {
        }
        //将线程构造成一个Node，添加到等待队列
        Node(Thread thread, Node mode) {     // Used by addWaiter
            this.nextWaiter = mode;
            this.thread = thread;
        }
        //这个方法会在Condition队列使用
        Node(Thread thread, int waitStatus) {
            this.waitStatus = waitStatus;
            this.thread = thread;
        }
    }

state的作用

　　state用来表示当前的同步状态，根据当前state的值，来判断当前释放处于锁定状态，或者是其他状态。而state的每一个值具体是什么含义，是由我们自己实现的。我们继承AQS时，根据自己的需求，实现一些方法，其中就是通过修改state的值来维持同步状态。而关于state，主要有以下三个方法：

//获取当前同步状态state的值
int getState()
//设置当前同步状态state的值
void setState(int newState)
//使用CAS设置当前同步状态的值，方法能够保证设置同步状态时的原子性；
//参数expect为state的预期旧值，而update是需要修改的新值，若设置成功，方法返回true，否则false
boolean compareAndSetState(int expect, int update)

AQS提供的模板方法

AQS提供的模板方法主要分为三类：

　　（1）独占式地获取和释放锁

　　（2）共享式地获取和释放锁

　　（3）查询AQS的同步队列中正在等待的线程情况

AQS可重写的方法

以上这些方法将会在AQS的模板方法中被调用，我们根据自己的需求，重写上述方法，控制同步状态state的值，即可控制线程同步的方式。

三、释放锁以及添加线程对于队列的变化

1、添加节点

　　当出现锁竞争以及释放锁的时候，AQS同步队列中的节点会发生变化，首先看一下添加节点的场景。

　　这里会涉及到两个变化：

1）新的线程封装成Node节点追加到同步队列中，设置prev节点以及修改当前节点的前置节点的next节点指向自己

2）通过CAS讲tail重新指向新的尾部节点

2、释放锁移除节点

　　head节点表示获取锁成功的节点，当头结点在释放同步状态时，会唤醒后继节点，如果后继节点获得锁成功，会把自己设置为头结点，节点的变化过程如下

　　这个过程也是涉及到两个变化：

1）修改head节点指向下一个获得锁的节点

2）新的获得锁的节点，将prev的指针指向null

　　这里有一个小的变化，就是设置head节点不需要用CAS，原因是设置head节点是由获得锁的线程来完成的，而同步锁只能由一个线程获得，所以不需要CAS保证，只需要把head节点设置为原首节点的后继节点，并且断开原head节点的next引用即可。

四、AQS源码分析

1、独占锁的实现原理

获取锁

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

1）通过tryAcquire尝试获取独占锁，如果成功返回true，失败返回false。

2）如果tryAcquire失败，则会通过addWaiter方法将当前线程封装成Node添加到AQS队列尾部。

3）执行acquireQueued方法，将Node作为参数，通过自旋去尝试获取锁。

4）成功获取锁后，将从acquireQueued方法中退出，同时返回一个boolean值表示当前线程是否被中断，若被中断，则会执行下面的selfInterrupt方法，响应中断。

tryAcquire(arg)

protected boolean tryAcquire(int arg) {
  throw new UnsupportedOperationException();
}

需要子类去实现，例如ReentrantLock等，作用就是尝试修改state值，也就是获取锁，若修改成功，则返回true，否则返回false。

addWaiter

当tryAcquire方法获取锁失败以后，则会先调用addWaiter将当前线程封装成Node，然后添加到AQS队列

// 将线程封装成一个节点，放入同步队列的尾部
private Node addWaiter(Node mode) {
    // 当前线程封装成同步队列的一个节点Node
    Node node = new Node(Thread.currentThread(), mode);
    // 这个节点需要插入到原尾节点的后面，所以我们在这里先记下原来的尾节点
    Node pred = tail;
    // 判断尾节点是否为空，若为空表示队列中还没有节点，则不执行以下步骤
    if (pred != null) {
        // 记录新节点的前一个节点为原尾节点
        node.prev = pred;
        // 将新节点设置为新尾节点，使用CAS操作保证了原子性
        if (compareAndSetTail(pred, node)) {
            // 若设置成功，则让原来的尾节点的next指向新尾节点
            pred.next = node;
            return node;
        }
    }
    // 若以上操作失败，则调用enq方法继续尝试
    enq(node);
    return node;
} 

enq(node)就是通过自旋操作把当前节点加入到队列中

private Node enq(final Node node) {
        //自旋
        for (;;) {
            Node t = tail; //如果是第一次添加到队列，那么tail=null
            if (t == null) { 
                //CAS的方式创建一个空的Node作为头结点
                if (compareAndSetHead(new Node()))
                   //此时队列中只一个头节点，所以tail也指向它
                    tail = head;
            } else {
　　　　　　　　//进行第二次循环时，tail不为null，进入else区域。
                //将当前线程的Node节点的prev指向tail，然后使用CAS将tail指向Node
                node.prev = t;
                if (compareAndSetTail(t, node)) {
　　　　　　　　　　//t此时指向tail,所以CAS成功后，将tail重新指向Node。
                    //此时t为更新前的tail的值，即指向空的头节点，t.next=node，就将头节点的后续结点指向Node，返回头节点
                    t.next = node;
                    return t;
                }
            }
        }
    }

将新线程封装成一个节点，加入到同步队列的尾部，若同步队列为空，则先在其中加入一个默认的节点，再进行加入；若加入失败，则自旋不断尝试，直到成功为止。这个过程中使用CAS保证了添加节点的原子性。

acquireQueued

将添加到队列中的Node作为参数传入acquireQueued方法，这里面会做抢占锁的操作

/**
 * 让线程不间断地获取锁，若线程对应的节点不是头节点的下一个节点，则会进入等待状态
 * @param node the node
 */
final boolean acquireQueued(final Node node, int arg) {
    // 记录失败标志
    boolean failed = true;
    try {
        // 记录中断标志，初始为true
        boolean interrupted = false;
        // 循环执行，因为线程在被唤醒后，可能再次获取锁失败，需要重写进入等待
        for (;;) {
            // 获取当前线程节点的前一个节点
            final Node p = node.predecessor();
            // 若前一个节点是头节点，则tryAcquire尝试获取锁，若获取成功，则执行if中的代码
            if (p == head && tryAcquire(arg)) {
                // 将当前节点设置为头节点
                setHead(node);
                // 将原来的头节点移出同步队列
                p.next = null; // help GC
                // 失败标志置为false
                failed = false;
                // 返回中断标志，acquire方法可以根据返回的中断标志，判断当前线程是否被中断
                return interrupted;
            }
            // shouldParkAfterFailedAcquire方法判断当前线程是否能够进入等待状态，
            // 若当前线程的节点不是头节点的下一个节点，则需要进入等待状态，
            // 在此方法内部，当前线程会找到它的前驱节点中，第一个还在正常等待或执行的节点，
            // 让其作为自己的直接前驱，然后在需要时将自己唤醒（因为其中有些线程可能被中断），
            // 若找到，则返回true，表示自己可以进入等待状态了；
            // 则继续调用parkAndCheckInterrupt方法，当前线程在这个方法中等待，
            // 直到被其他线程唤醒，或者被中断后返回，返回时将返回一个boolean值，
            // 表示这个线程是否被中断，若为true，则将执行下面一行代码，将中断标志置为true
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        // 上面代码中只有一个return语句，且return的前一句就是failed = false;
        // 所以只有当异常发生时，failed才会保持true的状态运行到此处；
        // 异常可能是线程被中断，也可能是其他方法中的异常，
        // 比如我们自己实现的tryAcquire方法
        // 此时将取消线程获取锁的动作，将它从同步队列中移除
        if (failed)
            cancelAcquire(node);
    }
}

1）获取当前节点的prev节点

2）如果prev节点为head节点，那么它就有资格去争抢锁，调用tryAcquire抢占锁

3）抢占锁成功以后，把获得锁的节点设置为head，并且移除原来的初始化head节点

4）如果获得锁失败，则根据waitStatus决定是否需要挂起线程

5）最后，通过cancelAcquire取消获得锁的操作

前面的逻辑都很好理解，主要看一下shouldParkAfterFailedAcquire这个方法和parkAndCheckInterrupt的作用

shouldParkAfterFailedAcquire

从上面的分析可以看出，只有队列的第二个节点可以有机会争取锁，如果成功获取锁，则此节点晋升为头节点。对于第三个及以后的节点，if (p == head)条件不成立，首先进行shouldParkAfterFailedAcquire(p, node)操作，shouldParkAfterFailedAcquire方法是判断一个争取锁的线程是否应该被阻塞。

它首先判断一个节点的前置节点的状态是否为Node.SIGNAL，如果是，是说明此节点已经将状态设置。如果锁释放，则应当通知它，所以它可以安全的阻塞了，返回true。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus; //前继节点的状态
    if (ws == Node.SIGNAL)//如果是SIGNAL状态，意味着当前线程需要被unpark唤醒
               return true;
　　 //如果前节点的状态大于0，即为CANCELLED状态时，则会从前节点开始逐步循环找到一个没有被“CANCELLED”节点设置为当前节点的前节点，返回false。　　 
    //在下次循环执行shouldParkAfterFailedAcquire时，返回true。这个操作实际是把队列中CANCELLED的节点剔除掉。
    if (ws > 0) {// 如果前继节点是“取消”状态，则设置 “当前节点”的 “当前前继节点” 为 “‘原前继节点'的前继节点”。
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else { // 如果前继节点为“0”或者“共享锁”状态，则设置前继节点为SIGNAL状态。
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

parkAndCheckInterrupt

如果shouldParkAfterFailedAcquire返回了true，则会执行parkAndCheckInterrupt()方法，它是通过LockSupport.park(this)将当前线程挂起到WATING状态，它需要等待一个中断、unpark方法来唤醒它，通过这样一种FIFO的机制的等待，来实现了Lock的操作。

private final boolean parkAndCheckInterrupt() {
  LockSupport.park(this);
  return Thread.interrupted();
}

LockSupport

LockSupport类是Java6引入的一个类，提供了基本的线程同步原语。LockSupport实际上是调用了Unsafe类里的函数，归结到Unsafe里，只有两个函数：

public native void unpark(Thread jthread);  
public native void park(boolean isAbsolute, long time); 

unpark函数为线程提供“许可(permit)”，线程调用park函数则等待“许可”。这个有点像信号量，但是这个“许可”是不能叠加的，“许可”是一次性的。

permit相当于0/1的开关，默认是0，调用一次unpark就加1，变成了1。调用一次park会消费permit，又会变成0。 如果再调用一次park会阻塞，因为permit已经是0了。直到permit变成1，这时调用unpark会把permit设置为1。每个线程都有一个相关的permit，permit最多只有一个，重复调用unpark不会累积。

释放锁

加锁的过程分析完以后，再来分析一下释放锁的过程，调用release方法，这个方法里面做两件事：释放锁、唤醒park的线程。

public final boolean release(int arg) {
    // 调用tryRelease尝试修改state释放锁，若成功，将返回true，否则false
    if (tryRelease(arg)) {
        // 若修改state成功，则表示释放锁成功，需要将当前线程移出同步队列
        // 当前线程在同步队列中的节点就是head，所以此处记录head
        Node h = head;
        // 若head不是null，且waitStatus不为0，表示它是一个装有线程的正常节点，
        // 在之前提到的addWaiter方法中，若同步队列为空，则会创建一个默认的节点放入head
        // 这个默认的节点不包含线程，它的waitStatus就是0，所以不能释放锁
        if (h != null && h.waitStatus != 0)
            // 若head是一个正常的节点，则调用unparkSuccessor唤醒它的下一个节点所对应的线程
            unparkSuccessor(h);
        // 释放成功
        return true;
    }
    // 释放锁失败
    return false;
}

tryRelease也是由子类具体去实现的，我们主要分析一下unparkSuccessor方法。在方法unparkSuccessor(Node)中，就意味着真正要释放锁了，它传入的是head节点（head节点是占用锁的节点），当前线程被释放之后，需要唤醒下一个节点的线程

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)java
        compareAndSetWaitStatus(node, ws, 0);
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {//判断后继节点是否为空或者是否是取消状态,
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
           //然后从队列尾部向前遍历找到最前面的一个waitStatus小于0的节点, 
　　　　　　 //至于为什么从尾部开始向前遍历，因为在doAcquireInterruptibly.cancelAcquire方法的处理过程中只设置了next的变化，没有设置prev的变化，
　　　　　　 //在最后有这样一行代码：node.next = node，如果这时执行了unparkSuccessor方法，并且向后遍历的话，就成了死循环了，所以这时只有prev是稳定的
        　　if (t.waitStatus <= 0)                 
           　　s = t;
    }
    //内部首先会发生的动作是获取head节点的next节点，如果获取到的节点不为空，则直接通过：“LockSupport.unpark()”方法来释放对应的被挂起的线程，    
    //这样一来将会有一个节点唤醒后继续进入循环进一步尝试tryAcquire()方法来获取锁
    if (s != null)
        LockSupport.unpark(s.thread); //释放许可
}

2、共享锁的实现原理

获取锁

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

可以看到，这个方法比较简短。首先调用tryAcquireShared方法尝试获取一次共享锁，即修改state的值，若返回值>=0，则表示获取成功，线程不受影响，继续向下执行；若返回值小于0，表示获取共享锁失败，则线程需要进入到同步队列中等待，调用doAcquireShared方法。acquireShared方法也是AQS的一个模板方法，而其中的tryAcquireShared方法就是需要使用者自己实现的方法。下面我们来看看doAcquireShared方法的实现

/**
 * 不间断地获取共享锁，若线程对应的节点不是头节点的下一个节点，将进入等待状态
 * 实现与acquireQueued非常类似*/
private void doAcquireShared(int arg) {
    // 往同步队列的尾部添加一个默认节点，Node.SHARED是一个Node常量，
    // 它的值就是一个不带任何参数的Node对象，也就是new Node();
    final Node node = addWaiter(Node.SHARED);
    // 失败标志，默认为true
    boolean failed = true;
    try {
        // 中断标志，用来判断线程在等待的过程中释放被中断
        boolean interrupted = false;
        // 死循环不断尝试获取共享锁
        for (;;) {
            // 获取默认节点的前一个节点
            final Node p = node.predecessor();
            // 判断当前节点的前一个节点是否为head节点
            if (p == head) {
                // 尝试获取共享锁
                int r = tryAcquireShared(arg);
                // 若r>0，表示获取成功
                if (r >= 0) {
                    // 当前线程获取锁成功后，调用setHeadAndPropagate方法将当前线程设置为head
                    // 同时，若共享锁还能被其他线程获取，则在这个方法中也会向后传递，唤醒后面的线程
                    setHeadAndPropagate(node, r);
                    // 将原来的head的next置为null
                    p.next = null; // help GC
                    // 判断当前线程是否中断，若被中断，则调用selfInterrupt方法响应中断
                    if (interrupted)
                        selfInterrupt();
                    // 失败标志置为false
                    failed = false;
                    return;
                }
            }
            // 以下代码和获取独占锁的acquireQueued方法相同，即让当前线程进入等待状态
            // 具体解析可以看上面acquireQueued方法的解析
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

doAcquireShared方法的实现和获取独占锁中的acquireQueued方法很类似，但是主要有一点不同，那就是线程在被唤醒后，若成功获取到了共享锁，还需要判断共享锁是否还能被其他线程获取，若可以，则继续向后唤醒它的下一个节点对应的线程。

释放锁

public final boolean releaseShared(int arg) {
    // 尝试修改state的值释放锁
    if (tryReleaseShared(arg)) {
        // 若成功，则调用以下方法唤醒后继节点中的线程
        doReleaseShared();
        return true;
    }
    return false;
}

releaseShared也是一个模板方法，它通过调用使用者自己实现的tryReleaseShared方法尝试释放锁，修改state的值，若返回true，表示修改成功，则继续向下调用doReleaseShared唤醒head的下一个节点对应的线程，让它开始尝试获取锁；若修改state失败，则返回false。

六、总结

　　在获得同步锁时，同步器维护一个同步队列，获取状态失败的线程都会被加入到队列中并在队列中进行自旋；移出队列（或停止自旋）的条件是前驱节点为头节点且成功获取了同步状态。在释放同步状态时，同步器调用tryRelease(int arg)方法释放同步状态，然后唤醒头节点的后继节点。