大厂面试必会AQS1——从ReentrantLock源码认识AQS

一、什么是AQS

AQS初识：

AbstractQuenedSynchronizer抽象的队列式同步器（简称队列同步器）。是除了java自带的synchronized关键字之外的锁机制。使用了一个int型的成员变量表示同步状态，子类继承同步器并实现它的抽象方法来管理同步状态，实现getState（）、SetState（int newState）、compareAndSetState（int expect，int update）（compareAndSetState的核心思想是CAS，对CAS不了解的请先观看《原子操作类AtomicInteger详解——由valatile引发的思考CAS初体验》）三个方法来实现状态的更改，通过内置的FIFO队列完成资源获取线程的排队工作。检查java中提供的lock类，我们发现如ReentrantLock，ReentrantReadWriteLock，StampedLock，CountDownLatch，CyclicBarrier等，内部都有AQS的实现类，完成了不同逻辑来承载不同Lock的实现。

AQS的核心思想

如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并将共享资源设置为锁定状态，如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制，这个机制AQS是用CLH队列锁实现的，即将暂时获取不到锁的线程加入到队列中。

CLH锁模型

是一种基于单向链表的、高性能、公平的自旋锁。申请加锁的线程通过前驱节点（pre-node）的变量进行自旋。当pre-node解锁后，当前节点会结束自旋并进行加锁。

1.locked == true 表示节点处于加锁状态或者等待加锁状态。
2. locked == false 表示节点处于解锁状态。
3. 基于线程当前节点的前置节点的锁值（locked）进行自旋，前置节点的 locked == true 自旋；当前置节点解锁时，设置locked == false，后继节点（就是当前节点）监听到false，结束自旋。
4. 每个节点在解锁时更新自己的锁值（locked），在这一时刻，该节点的后置节点会结束自旋，并进行加锁。

由于自旋过程中，监控的是前置节点的变量，因此在SMP架构的共享内存模式，能更好的提供性能

MCS锁模型

与CLH锁模型的最大区别是，监控的是自己的节点变量，当前置节点解锁后，会主动修改自己的节点变量状态。这种模型解决的是CLH模型在NUMA架构上的不足：当前置节点存在于其他CPU模块时，自旋会导致频繁的调用互联模块。是将自旋调整到了节点自身，互联模块的调用只存在于前置节点解锁的时刻。

1.locked == false 标识节点处于加锁状态（没有自旋）
2. locked == true 标识节点处于等待状态（自旋）
3. 基于当前节点的锁值（locked）进行自旋，locked == true 自旋；当前置节点解锁时，修改后继节点（就是当前节点）的 locked == false ，进而结束当前节点的自旋。
4.每个节点在解锁时更新后继节点的锁值（locked），在这一刻，该节点的后置节点会结束自旋，并进行加锁。

二、AQS的实现分析

同步队列

前面说到同步器依赖同步队列管理同步状态，当线程获取同步状态失败时，会将当前线程以及等待状态信息构造成一个Node节点加入到同步队列，同时阻塞当前线程，当同步状态释放时，会唤醒首节点的线程使其再次尝试获取同步状态，同步队列的基本结构为：

在这里插入图片描述

可以看到同步队列其实是一个虚拟的双向链表，由数个Node组成，head节点没有prev前驱节点，而tail尾结点没有next后继节点，剩下的每一个Node节点都有自己的前驱和后继节

Node用来保存获取同步状态失败的线程的引用、等待状态以及前驱和后继节点
参考《java并发编程的艺术》节点属性描述为：

在这里插入图片描述

同步器提供的方法列表：

在这里插入图片描述

模板方法基本分为三类：独占式同步状态获取与释放、共享式同步状态获取与释放和查询同步队列中等待线程情况。

三从ReentrantLock源码认识AQS

首先是lock方法，以公平锁为例

public void lock() {
    sync.lock();
}
final void lock() {
    acquire(1);
}

稍微提一下非公平锁

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

可以看到区别在于非公平锁的实现会先尝试占有锁，失败后在以公平锁的方式获取锁。下面分析一下acquire，独占式同步状态的获取

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

可以看到首先tryAcquire方法

static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    /**
     * Fair version of tryAcquire.  Don't grant access unless
     * recursive call or no waiters or is first.
     */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        //加锁次数
        int c = getState();
        if (c == 0) {
        //如果队列没有线程在前面排队，CAS更新State由0到1，将当前线程设置为独占锁的拥有者，与公平锁的实现方式区别于多了个hasQueuedPredecessors方法的判断，需要排队
            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;
    }
}

如果tryAcquire（）返回false，即获取同步状态失败，将当前线程以及等待状态信息构造成一个Node节点加入到同步队列，然后中断当前线程，接下来看addWaiter(Node.EXCLUSIVE)

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) { //如果尾结点为空，初始化一个头结点，并将尾结点指向头结点
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {//尾结点非空，将传入的节点插入尾部
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

enq方法如果尾结点为空，会先初始化一个头结点（也叫哨兵节点，初始状态位为0），并将尾结点指向头结点，接下来继续循环，此时尾结点不为空，则addWaiter方法返回的节点插入尾部

最后调用acquireQueued(Node node,int arg)方法，使得该节点以“死循环”的方式获取同步状态。如果获取不到阻塞节点中的线程，而被阻塞线程的唤醒主要依靠前驱节点的出队或阻塞线程被中断来实现。

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();//前驱节点
            //如果前驱节点是头结点并且tryAcquire（）获取同步状态
            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);
    }
}

可以看到acquireQueued的主要作用是把已经追加到队列的线程节点（addWaiter方法返回值）进行阻塞，但阻塞前如果前驱节点是头结点的话还是有机会再通过tryAccquire重试来获得锁，如果重试成功则无需阻塞，直接返回

但是如果p == head && tryAcquire(arg)条件不满足循环将永远无法结束，就要看下面的代码

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.
         */
         //如果前继的节点状态为SIGNAL(前驱结点释放时会唤醒后继节点)，表明当前节点需要park，则返回成功，此时acquireQueued方法中（parkAndCheckInterrupt）将导致线程阻塞规则
        return true;
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
         //如果前继节点状态为CANCELLED(ws>0)，说明前置节点已经被放弃，则回溯到一个非取消的前继节点，返回false，acquireQueued方法的无限循环将递归调用该方法，直至规则1返回true，导致线程阻塞规则
        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.
         */
         //如果前继节点状态为非SIGNAL、非CANCELLED，则设置前继的状态为SIGNAL（保证自己阻塞后可以被前驱结点唤醒)，返回false后进入acquireQueued的无限循环
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

举例：假设A线程已经获得锁，B线程获得锁失败将会使它的前一个节点（状态从0变为-1）shouldParkAfterFailedAcquire方法的else分支，然后在循环体中再次进入该方法，返回ture，则继续执行parkAndCheckInterrupt方法。

这代表着，同步队列里面只有tail节点状态是0，其余节点都被在下一个节点加入同步队列的时候状态变为-1

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

调用LockSupport.park最终把线程交给系统（Linux）内核进行阻塞，即线程B阻塞在这个地方，这样acquireQueued就不会真正死循环了。

那么线程B如何被唤醒呢？且看unlock解锁是调用了release方法

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

protected final boolean tryRelease(int releases) {
//加锁次数-1
    int c = getState() - releases;
    //当前线程不是获得锁的线程，抛异常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    //可重入锁每解锁一次c= getState() - releases,当c == 0，代码解锁完毕
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

我们可以看到，当该方法执行时，如果线程多次锁定，则进行多次释放，直至status==0则真正释放锁，所谓释放锁即设置status为0，因为无竞争所以没有使用CAS。release的语义在于：如果可以释放锁，则唤醒队列第一个线程（Head）

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;
    //如果后继节点为null或者被取消，继续寻找下一个直到找到可用的后继节点
    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);
}

该方法中node为头节点，s为B线程节点
可以看到unparkSuccessor方法会找到一个没有被取消的后继节点并将其唤醒
若B线程状态不是取消状态，则B线程被唤醒并从上文阻塞处开始重新竞争锁。

总结：
　　　　
lock方法：

如果是公平锁，直接调用acquire(1);方法；如果是非公平锁先cas尝试获取锁，如果成功，将exclusiveOwnerThread置为当前线程，
acquire（1）首先执行tryAcquire方法，尝试获取锁，如果锁状态是0代码没有其他线程占用，则获取锁，将锁状态置1，exclusiveOwnerThread置为当前线程；如果锁状态不为1，就要看获取锁的线程是否是当前线程，如果是，锁状态+1，这里是可重入锁的关键，否则tryAcquire失败返回false
tryAcquire如果返回false，代表尝试是失败，此时要将当前线程放入同步队列：首先执行addWaiter(Node.EXCLUSIVE)方法，此方法会创建一个node节点，如果同步队列为空，enq方法中第一次循环时初始化队列并创建一个哨兵节点head节点状态位0，第二次循环时将创建新的节点排在head节点之后。
最后执行acquireQueued方法，此时如果当前节点的前驱结点是head节点，还有一次机会调用tryAcquire获取锁，如果仍然失败，则会shouldParkAfterFailedAcquire方法以保证前驱结点的waitStatus为SIGNAL（-1，这样的节点在释放锁时会唤醒一个可用的后继节点），这样就可以调用parkAndCheckInterrupt方法阻塞当前线程以待其它线程将其唤醒
shouldParkAfterFailedAcquire方法的规则使得队列中除了tail节点的所有节点状态为-1，而tail节点为0.

unlock方法

调用sync.release(1)方法，首先调用tryRelease方法尝试解锁，此方法会将同步状态-1，但必须保证获取锁的线程是当前线程，否则抛出异常。如果同步状态-1后为0，则代表锁释放完毕，将获取锁得线程置null，否则，代表可重入锁没有释放完毕，仅仅只需将同步状态-1
释放锁以后，如果同步队列存在没有被取消的后继节点，则唤醒它
被唤醒的线程继续此时如果前驱节点是head节点，则继续tryAcquire获得锁，如果成功，当前节点会替换原来的head节点成为新的head节点，也即哨兵节点

	private void setHead(Node node) {
	        head = node;
	        node.thread = null;
	        node.prev = null;
	    }

unpark和park方法只有一个通行证，多次unpark也不会累加，即一次park就会消耗掉通行证，无论之前unpark多少次

接下来会继续更新AQS的共享式同步状态的获取与释放请大家期待

大厂面试必会AQS1——从ReentrantLock源码认识AQS

一、什么是AQS

二、AQS的实现分析

三 从ReentrantLock源码认识AQS

相关推荐

三从ReentrantLock源码认识AQS