基于AQS实现锁机制需要关心什么？

AQS由一个FIFO的双向队列以及一个单一的状态信息state组成；对于state，AQS提供了getState()、setState()、compareAndSetState(int expect, int update)三种方法用来对state进行操作；

每个基于AQS实现的锁机制，对于state都有不同的定义：

• ReentrantLock作为排他锁，对于state状态定义为：state == 0 时表示当前ReentrantLock锁状态为空闲，state >= 1时表示当前有线程持有锁，并且表示锁的重入次数；
• semaphore，信号量作为共享锁，对于state状态定义为：state > 0时表示当前仍然存在可以获得的资源，state <= 0 时表示已经不存在可以获取的资源了；
• ReentrantReadWriteLock作为读写锁(排他共享锁)，具有读锁共享，写锁排他但可重入的性质，对于state状态的定义为：state的高16位表示持有ReentrantReadWriteLock读锁的线程数；第16位表示持有ReentrantReadWriteLock写锁线程的可重入次数；

可见不同的锁机制按照锁本身获取与释放的语义，定义了state所处的不同状态值下锁资源当前是否可获取的状态； 因此不同的锁机制要按照自身锁语义实现不同锁的获取与释放的条件；

• 排他锁需要覆盖实现boolean tryAcquire(int arg)、boolean tryRelease(int arg)、boolean isHeldExclusively()定义锁自身的锁获取与释放的语义；
• 共享锁需要覆盖实现int tryAcquireShared(int arg)、boolean tryReleaseShared(int arg)定义锁自身的锁获取与释放的语义；

排他锁ReentrantLock：

获取锁

通过线程获取ReentrantLock锁的语义：1)排他性 2)可重入性， 定义对应的tryAcquire()方法：

// state状态值
private volatile int state;

// ReentrantLock tryAcquire的非公平实现
// 此处的非公平是指未进入AQS阻塞队列的线程抢占式获取锁，无需关心阻塞队列中是否有其他线程等待锁；
// 而之前已经抢占锁失败进入阻塞队列中的线程仍然需要按照出队列的顺序依次获取锁。
// 这个点之后讲AQS原理时会细讲，这里先提醒一下，防止对非公平性存在先入为主的观念。
protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        // 获取当前state
        int c = getState();
        // 当前state值为0，说明当前锁处于空闲状态，因此当前请求获取锁的线程可以尝试CAS获取锁
        if (c == 0) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                // 返回true表示当前线程按照锁获取的语义可以得到锁
                return true;
            }
        }
        // state值不为0，说明当前锁已经被某个线程持有，
        // 如果持有锁的线程正是当前线程，则增加state的值，表示锁的重入次数
        else if (current == getExclusiveOwnerThread()) {
            // 此处请求的资源量(acquires = 1)，即锁的重入次数 + 1
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            // 修改state的状态值
            setState(nextc);
            // reentrantLock支持可重入，因此当前线程仍然可以获取到锁
            return true;
        }
        // 都不满足，则当前线程获取锁失败，交由AQS阻塞队列管理
        return false;
}
复制代码

释放锁

通过持有锁的线程释放锁的语义：1)可重入性->对于多次重入的锁要多次释放 2）排他性 -> 只有持有锁的线程才可以释放锁，否则说明该线程持有锁期间发生了线程安全问题，需要抛出异常。

protected final boolean tryRelease(int releases) {
    // 当前state值 - 要释放的资源
    int c = getState() - releases;
    // 只有持有锁的线程才可以释放锁
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    // 只有释放资源后state值变为0，才可以释放锁
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}
复制代码

读写锁ReentrantReadWriteLock

获取读锁

通过线程获取ReentrantReadWriteLock读锁的语义：1) 共享锁 -> 读锁可以被多个线程共享 2)读写互斥 -> 当前有线程获取了写锁，则获取读锁必然失败

// 获取int的低16位值 -> 即持有写锁线程的重入次数
static int exclusiveCount(int c) { return c & 0x00FF; }
// 获取int的高16位值 -> 即持有读锁线程的数量（最多为2^16-1，否则int值溢出）
static int sharedCount(int c)    { return c >>> 16; }

protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
    // 获取当前state状态
    int c = getState();
    // 当前state的低16位不为0，即有线程获取了写锁，并且写锁不是当前线程所持有则获取读锁失败
    // PS：已经获取写锁的线程是可以获取读锁的
    if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) {
        return -1;
    }
    // 获取持有读锁线程的数量
    int r = sharedCount(c);
    /*
     * readerShouldBlock()方法依据公平与非公平的算法提供了不同的实现
     * 公平：阻塞队列中有线程时，只有请求线程为head元素时才可以获取到读锁；
     *       阻塞队列为空时，请求线程可以获取锁。
     * 非公平：只要调用`tryAcquireShared`获取读锁的线程都可以抢占式获取锁。
     */
     // 如果该请求线程拥有获取锁的权力，并且当前读锁数量不超过2^16-1
     // 则可以通过CAS对state高16位值加一
    if (!readerShouldBlock() && r < MAX_COUNT 
                             && compareAndSetState(c, c + SHARED_UNIT)) {
        // 省略通过ThreadLocal维护每个线程读锁重入次数的代码
        ...
        // 获取读锁成功
        return 1;
    }
}
复制代码

释放读锁：

释放读锁时，因为读锁是共享锁，所以并不会有影响；但是如果当前读锁释放后，读锁持有量变为0，则可以尝试优先唤醒写线程，防止写线程对锁饥饿；

protected final boolean tryReleaseShared(int unused) {
    // 省略对于读锁重入次数的维护
    ...
    // 死循环+CAS确保释放当前线程读锁
    for (;;) {
        int c = getState();
        int nextc = c - SHARED_UNIT;
        if (compareAndSetState(c, nextc))
            // 如果当前读锁释放后读锁的持有量变为0，下层组件AQS会释放一个因为获取写锁而阻塞的线程
            return nextc == 0;
    }
}
复制代码

总结

因此锁的实现者基于AQS去实现锁时，关注的核心重点在于要实现的锁自身的锁语义，并根据锁语义定义state状态的含义以及获取锁、释放锁时如何修改state的状态； 但是作为锁的实现者，仍然需要关心AQS的内部细节，更好地利用AQS提供的机制，比如读写锁中当读锁持有量变为0时return true可以一定程度上缓解写锁饥饿的问题；

下面文章的核心重点在：AQS在背后做了什么。

AQS做了什么？

基础概念

可以先留个印象，这几个概念会贯穿下文：

Node对象的waitStatus属性

// 该waitStatus值表示当前Node对应的线程已经被取消
static final int CANCELLED =  1;
// 该waitStatus值表示当前Node释放锁时需要唤醒阻塞队列中下一个节点
static final int SIGNAL    = -1;
// 该waitStatus值表示当前Node节点阻塞在相应的条件(condition)上
static final int CONDITION = -2;
// 表示下一次共享式同步状态获取将会无条件被传播下去
static final int PROPAGATE = -3;
复制代码

LockSupport工具类

该工具类封装了UnSafe类，提供了挂起和唤醒线程操作的能力，即park与unpark系列方法；

处理中断的方式

wait系列、join()、sleep()等基础方法同样可以挂起线程，如线程A通过该类方法挂起后，如果其他线程通过interrupt()方法设置了线程A的中断标志位，那么线程A会抛出InterruptedException异常并清空线程的中断状态；

而通过调用park()方法而被阻塞的线程被其他线程中断而返回时并不会抛出InterruptedException异常，也不会清空线程的中断状态；

也是基于此特性，AQS提供了响应中断与不响应中断的两套方法。

阻塞线程的唤醒条件

调用park()方法阻塞的线程，在三种情况下会返回：(1) 其他线程调用了unpark()方法唤醒该线程；(2)其他线程调用interrupt()中断了该线程；(3)虚假唤醒。

因此在下面分析源码时，当被阻塞的线程唤醒后，都会重新判断一下中断标志位，来区分当前线程是通过unpark()还是interrupt唤醒的，如果是通过中断唤醒，需要判断是否抛出异常。

AQS阻塞队列

AQS通过模板模式封装了一套获取锁、释放锁流程的算法；

获取独占锁流程

其中acquire(int arg)方法定义了获取锁的流程：

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer {
    // 注意 EXCLUSIVE值为null，会在下面addWaiter()方法中讲到设置为null的原因
    static final Node EXCLUSIVE = null;
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
}
复制代码

1. 判断判断当前线程是否可以获取锁：通过调用锁实现者通过继承AbstractingQueueSychronizer抽象类实现的抽象方法-tryAcquire(int arg)；
2. 获取锁成功：该方法直接返回，表明获取锁成功；
3. 获取锁失败：即tryAcquire()返回false，因此会接着依次调用addWaiter(Node.EXCLUSIVE)与acquireQueued()方法；其中addWaiter()负责新建当前线程对应的Node节点，并通过CAS+死循环的方式将该节点放入AQS阻塞队列队尾；而acquireQueued()负责将获取锁失败的线程Node节点通过LockSupport.park()进行挂起，等待前置节点的唤醒；

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer {

    // 静态内部类-Node
    static final class Node {
        static final Node EXCLUSIVE = null;
        // 如果当前节点在某个condition单向队列上，那么该值为指向condition队列下一个Node的指针
        Node nextWaiter;
        /** Constructor used by addWaiter. */
        Node(Node nextWaiter) {
            this.nextWaiter = nextWaiter;
            // 将当前Node与当前线程关联起来
            THREAD.set(this, Thread.currentThread());
        }
    }    
    
    /**
     * 为当前线程创建Node，并放入AQS队列的队尾
     *
     * 如果该方法调用者是acquire()，那么此时mode为null，即创建的Node节点的nextWaiter为空指针；
     * 因为condition队列只有通过持有锁的线程调用signal方法才可以进入；
     * 这个下面讲condition时会具体讲
     */
    private Node addWaiter(Node mode) {
        Node node = new Node(mode);

        // 死循环 + CAS将当前队列插入队尾
        for (;;) {
            Node oldTail = tail;
            if (oldTail != null) {
                node.setPrevRelaxed(oldTail);
                if (compareAndSetTail(oldTail, node)) {
                    oldTail.next = node;
                    return node;
                }
            } else {
                initializeSyncQueue();
            }
        }
    }
    
    final boolean acquireQueued(final Node node, int arg) {
        boolean interrupted = false;
        try {
            for (;;) {
                final Node p = node.predecessor();
                // 如果当前节点的前置(predecessor)节点是首节点，并且获取锁成功；
                // 则将当前节点设置为首节点。
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    // 将原首节点的next设置为null，使得首节点不持有其他对象的引用；
                    // 按照JVM的可达性分析算法，该对象成为垃圾回收的目标，help GC
                    p.next = null; // help GC
                    return interrupted;
                }
                // 获取锁失败，则将当前节点的前置节点waitStatus设置为SIGNAL，并将自己挂起；
                // 设置前置节点为SIGNAL是为了在前置节点获取到锁并成为首节点后，
                // 通过release方法释放锁时会唤醒该节点
                if (shouldParkAfterFailedAcquire(p, node))
                    interrupted |= parkAndCheckInterrupt();
            }
        } catch (Throwable t) {
            cancelAcquire(node);
            if (interrupted)
                selfInterrupt();
            throw t;
        }
    }
}
复制代码

释放独占锁流程

AQS的release(int arg)方法定义了释放锁的流程：

1. 判断当前线程是否释放锁成功：通过调用锁实现者通过继承AbstractingQueueSychronizer抽象类实现的抽象方法-tryRelease(int arg)；
2. 释放成功时：通过unparkSuccessor(Node node)方法唤醒阻塞队列中下一个存活节点；
3. 释放失败时：因为独占锁释放一般不会失败，如果失败那一定是因为锁的实现者增加了释放的约束条件，所以AQS会交由锁实现者处理，

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer {
    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                // 唤醒下一个节点
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    
    /**
     * Wakes up node's successor, if one exists.
     *
     * @param node the node 从relase()方法调用时，node为阻塞队列首节点
     */
    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)
            node.compareAndSetWaitStatus(ws, 0);

        /*
         * 正常情况下，首节点会唤醒的节点为自身next指针指向的节点；
         * 但是存在下一个节点线程被CANCELED的情况（waitStatus > 0）
         * 因此该方法会从阻塞队列的尾部反向遍历节点，找到离node节点最近的一个有效节点：
         * 即首节点真正要唤醒的下一个节点。
         */
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node p = tail; p != node && p != null; p = p.prev)
                if (p.waitStatus <= 0)
                    s = p;
        }
        // 唤醒首节点真正要唤醒的下一个节点
        if (s != null)
            LockSupport.unpark(s.thread);
    }
}
复制代码

下一个线程被唤醒后

这里为了逻辑的闭环，再啰嗦一下； 线程节点是在acquireQueued()方法中通过parkAndCheckInterrupt()方法挂起的：

private final boolean parkAndCheckInterrupt() {
    // 阻塞线程卡在park方法处
    LockSupport.park(this);
    // 线程被唤醒后会接着获取线程挂起期间是否被其他线程通过interrupt()方法设置过中断标记后返回
    return Thread.interrupted();
}
复制代码

parkAndCheckInterrupt()方法返回后，线程会开始新的一轮for循环，并重新判断前置节点是否为head节点，并重新tryAcquire(arg)尝试获得锁；

for (;;) {
    final Node p = node.predecessor();
    if (p == head && tryAcquire(arg)) {
            setHead(node);
            p.next = null; // help GC
            return interrupted;
        }
    if (shouldParkAfterFailedAcquire(p, node))
            interrupted |= parkAndCheckInterrupt();
}
复制代码

此时如果锁的实现者使用了非公平性的策略，那么如果此时有尚未进入阻塞队列的其他线程调用acquire() -> tryAcquire()方法获取锁时，则当前该线程不一定获取到锁，需要比较时间的先后关系；

如果锁的实现者使用了公平性的策略，那么位于首节点后的该线程一定会获取到锁，因为尚未进入阻塞队列的其他线程调用tryAcquire()时会额外调用hasQueuedPredecessors()方法进行判定： 如果AQS阻塞队列中存在等待线程节点，则该线程获取锁失败后加入AQS阻塞队列； 如果AQS阻塞队列中不存在等待线程节点，并且当前state状态为可以获取锁状态，那么该线程获取到锁。

不过，无论锁的公平与否，只要线程进入了阻塞队列，那么所有的线程只可以按照队列先进先出的顺序依次尝试获得锁。

public final boolean hasQueuedPredecessors() {
    Node h, s;
    if ((h = head) != null) {
       if ((s = h.next) == null || s.waitStatus > 0) {
            s = null; // traverse in case of concurrent cancellation
            for (Node p = tail; p != h && p != null; p = p.prev) {
                if (p.waitStatus <= 0)
                    s = p;
            }
        }
        // 当前AQS阻塞队列中存在等待线程
        if (s != null && s.thread != Thread.currentThread())
            return true;
    }
    // 当前AQS阻塞队列中不存在等待线程
    return false;
}
复制代码

加入condition队列

为什么需要condition

在线程协作方面，内部锁的锁粒度是对象层面的，即如果一个类中有两个synchronized方法，那么同一时刻两个方法不能被同时访问，即使这两个方法并不构成逻辑冲突；并且内部锁使用notify()进行随机唤醒_WaitSet中的等待线程，可能会导致过早唤醒的问题，因为如果两个方法等待的条件并不一致时，可能通过notify()随机唤醒并获取对象访问权的线程并没有满足执行的条件。

因此在Java中新增了Condition类用作唤醒条件维度的抽象，为每一个唤醒条件维护了一个阻塞队列从而解决了这个问题。

如何构建condition队列

日常使用ReentrantLock时，只需要通过lock.newCondition即可，下面的代码便构建了俩个condition队列：

    static ReentrantLock produceAndConsumerLock = new ReentrantLock();
    static Condition bufferAbleToPut = produceAndConsumerLock.newCondition();
    static Condition bufferAbleToConsume = produceAndConsumerLock.newCondition();
复制代码

内部实现则依赖AQS中的ConditionObject对象，通过ConditionObject属性firstWaiter与lastWaiter作为队列的首尾指针，通过Node中nextWaiter指针连接队列元素构成单向链表；

因此在ReentrantLock中，只是很简单地构建一个ConditionObject便返回了：

    public Condition newCondition() {
        return sync.newCondition();
    }
    // 新建一个AQS内部类ConditionObject对象并返回
    abstract static class Sync extends AbstractQueuedSynchronizer {
        final ConditionObject newCondition() {
            return new ConditionObject();
        }
    }
复制代码

加入condition队列流程(await())

与只有获取内部锁进入Synchronized临界区的线程可以调用wait()、notify()、notifyAll()相同，基于AQS实现的锁机制，也只有持有当前锁的对象可以调用await()、signal()、signalAll()；因为通过锁的排他性，可以保证唤醒/挂起线程操作的原子、可见、有序，不会出现因为并发而导致的错误；

当获取到AQS锁的线程，判断当前的condtion条件不满足时，则会调用await()方法加入condition条件队列；

public class ConditionObject implements Condition, java.io.Serializable {
        
        public final void await() throws InterruptedException {
            // 响应中断时机之一：进入方法前检查中断标志位
            if (Thread.interrupted())
                throw new InterruptedException();
            // 新建一个Node节点(waitStatus=CONDITION)，并尾插至Condition队列
            Node node = addConditionWaiter();
            // 唤醒下一个AQS阻塞队列节点，原头节点被释放等待GC
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            // 判断当前Node节点是否在AQS队列中，如果不在则挂起
            // （如果该Node被signal、signalAll方法唤醒，则会移出Condition队列，
            //   加入到AQS阻塞队列中）
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                // 如果当前Node节点对应的线程在挂起过程中被设置了中断，则跳出当前循环
                // 等待过程中线程被取消时，interruptMode被赋为THROW_IE，否则设为REINTERRUPT
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            // 从上面的while循环退出时，表明当前Node已经处于AQS队列中，因此尝试获取锁
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            // 响应中断时机之二：存在两种方式，THROW_IE：直接抛出IE异常 REINTERRUPT：自我中断
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }
}
复制代码

移出condition队列加入阻塞队列流程(signal())

持有AQS锁的线程，当发现满足condition唤醒条件时，则会进行唤醒：

public class ConditionObject implements Condition, java.io.Serializable {

        public final void signal() {
            // 只有持有AQS锁的线程才可以进行唤醒等待队列中的线程
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
           
            Node first = firstWaiter;
            // 阻塞队列中还存在元素时，唤醒首节点
            if (first != null)
                doSignal(first);
        }


        private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }
        
        
        final boolean transferForSignal(Node node) {
            // 如果当前节点的waitStatus已经不是CONDITION，则说明当前线程已经被取消了
            // 则上层doSignal()方法会尝试继续唤醒条件队列中下一个CONDITION节点
            if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
                return false;
            // 节点对应线程未被取消则进入AQS阻塞队列，返回node在AQS阻塞队列中的前置节点
            Node p = enq(node);
            int ws = p.waitStatus; // 前置节点的状态
            // 如果前置节点是被取消的无效节点，则直接唤醒
            // 如果前置节点不是无效节点，则将前置节点waitStatus设置为SIGNAL
            if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
                LockSupport.unpark(node.thread);
            return true;
    }
}
复制代码

进入阻塞队列后的流程就在await()的后半段，会通过acquireQueued()方法等待获取到AQS锁，不过此时该node对应的线程获取到AQS锁时，未必满足condition的条件，因为可能在该node从条件队列尾插阻塞队列前，已经有多个等待同样条件的线程排在阻塞队列中（它们还未获取过AQS锁，因此也没有加入到条件队列中）。