ReentrantLock 源码分析以及 AQS （一）

前言

jdk1.5 之后发布了juc（java.util.concurrent），用于解决多线程并发问题。aqs 是一个特别重要的同步框架，很多同步类都借助于 aqs 实现了对线程同步状态的管理。

aqs 中最主要的就是独占锁和共享锁的获取和释放，以及提供了一些可中断的获取锁，超时等待锁等方法。

reentranlock 是基于 aqs 独占锁的一个实现。reentrantreadwritelock 是基于 aqs 共享锁的一个读写锁实现。本来打算一篇文章里面写完独占锁和共享锁，但是发现篇幅太长了，也不易于消化。

因此，本篇就先结合 reentrantlock 源码，分析 aqs 的独占锁获取和释放。以及 reentrantlock 的公平锁和非公平锁实现。

下一篇再写 reentrantreadwritelock 读写锁源码，以及 aqs 共享锁的获取和释放。

在正式讲解源码之前，墙裂建议读者做一些准备工作，最好对以下知识有一定的了解，这样阅读起来源码会比较轻松（因为，我当初刚开始接触多线程时，直接看 aqs 简直是一脸懵逼，就像读天书一样。。）。

1. 了解双向链表的数据结构，以及队列的入队出队等操作。
2. locksupport 的 park，unpark 方法，以及对线程的 interrupt 几个方法了解（可参考：locksupport的 park 方法是怎么响应中断的？）。
3. 对 cas 和自旋机制有一定的了解。

aqs 同步队列

aqs 内部维护了一个 fifo（先进先出）的双向队列。它的内部是用双向链表来实现的，每个数据节点（node）中都包含了当前节点的线程信息，还有它的前后两个指针，分别指向前驱节点和后继节点。下边看一下 node 的属性和方法：

static final class node {
    //可以认为是一种标记，表明了这个 node 是以共享模式在同步队列中等待
    static final node shared = new node();
    //也是一种标记，表明这个 node 是以独占模式在同步队列中等待
    static final node exclusive = null;

    /** waitstatus 常量值 */
    //说明当前节点被取消，原因有可能是超时，或者被中断。
    //节点被取消的状态是不可逆的，也就是说此节点会一直停留在取消状态，不会转变。
    static final int cancelled =  1;
    //说明后继节点的线程被 park 阻塞，因此当前线程需要在释放锁或者被取消时，唤醒后继节点
    static final int signal    = -1;
    //说明线程在 condition 条件队列等待
    static final int condition = -2;
    //在共享模式中用，表明下一个共享线程应该无条件传播
    static final int propagate = -3;

    
    //当前线程的等待状态，除了以上四种值，还有一个值 0 为初始化状态（条件队列的节点除外）。
    //注意这个值修改时是通过 cas ，以保证线程安全。
    volatile int waitstatus;

    //前驱节点
    volatile node prev;

    //后继节点
    volatile node next;

    //当前节点中的线程，通过构造函数初始化，出队时会置空（这个后续说，重点强调）
    volatile thread thread;

    //有两种情况。1.在 condition 条件队列中的后一个节点 
    //2. 一个特殊值 shared 用于表明当前是共享模式（因为条件队列只存在于独占模式）
    node nextwaiter;

    //是否是共享模式，理由同上
    final boolean isshared() {
        return nextwaiter == shared;
    }

    //返回前驱节点，如果为空抛出空指针
    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;
    }
}

另外，在 aqs 类中，还会记录同步队列的头结点和尾结点：

    //同步队列的头结点，是懒加载的，即不会立即创建一个同步队列，
    //只有当某个线程获取不到锁，需要排队的时候，才会初始化头结点
    private transient volatile node head;

    //同步队列的尾结点，同样是懒加载。
    private transient volatile node tail;

独占锁

这部分就结合 reentrantlock 源码分析 aqs 的独占锁是怎样获得和释放锁的。

非公平锁

首先，我们从 reentrantlock 开始分析，它有两个构造方法，一个构造，可以传入一个 boolean 类型的参数，表明是用公平锁还是非公平锁模式。另一个构造方法，不传入任何参数，则默认用非公平锁。

public reentrantlock() {
    sync = new nonfairsync();
}

public reentrantlock(boolean fair) {
    sync = fair ? new fairsync() : new nonfairsync();
}

nonfairsync 和 fairsync 都继承自 sync ，它们都是 reentranlock 的内部类。 而sync 类又继承自 aqs (abstractqueuedsynchronizer)。

static final class nonfairsync extends sync {
}

static final class fairsync extends sync {
}

abstract static class sync extends abstractqueuedsynchronizer {
}

知道了它们之间的继承关系，我们就从非公平锁的加锁方法作为入口，跟踪源码。因为非公平锁的流程讲明白之后，公平锁大致流程都一样，只是多了一个条件判断（这个，一会儿后边细讲，会做对比）。

nonfairsync.lock

我们看下公平锁的获取锁的方法：

final void lock() {
    //通过 cas 操作把 state 设置为 1
    if (compareandsetstate(0, 1))
        //如果设值成功，说明加锁成功，保存当前获得锁的线程
        setexclusiveownerthread(thread.currentthread());
    else
        //如果加锁失败，则执行 aqs 的acquire 方法
        acquire(1);
}

public final void acquire(int arg) {
    if (!tryacquire(arg) &&
        acquirequeued(addwaiter(node.exclusive), arg))
        selfinterrupt();
}

acquire

这个方法的逻辑是：

1. 通过 tryacquire 方法，尝试获取锁，如果成功，则返回 true，失败返回 false 。
2. tryacquire 失败之后，会先调用 addwaiter 方法，把当前线程封装成 node 节点，加入同步队列（独占模式）。
3. acquirequeued 方法会把刚加入队列的 node 作为参数，通过自旋去获得锁。

tryacquire

这是一个模板方法，具体的实现需要看它的子类，这里对应的就是 reentrantlock.nonfairsync.tryacquire 方法。我们看一下：

protected final boolean tryacquire(int acquires) {
    return nonfairtryacquire(acquires);
}

final boolean nonfairtryacquire(int acquires) {
    //当前线程
    final thread current = thread.currentthread();
    //获取当前的同步状态，若为 0 ，表示无锁状态。若大于 0，表示已经有线程抢到了锁。
    int c = getstate();
    if (c == 0) {
        //然后通过 cas 操作把 state 的值改为 1。
        if (compareandsetstate(0, acquires)) {
            // cas 成功之后，保存当前获得锁的线程
            setexclusiveownerthread(current);
            return true;
        }
    }
    // 如果 state 大于0，则判断当前线程是否是获得锁的线程，是的话，可重入。
    else if (current == getexclusiveownerthread()) {
        //由于 reentrantlock 是可重入的，所以每重入一次 state 就加 1 。
        int nextc = c + acquires;
        if (nextc < 0)
            throw new error("maximum lock count exceeded");
        setstate(nextc);
        return true;
    }
    return false;
}

addwaiter

如果获取锁失败之后，就会调用 addwaiter 方法，把当前线程加入同步队列。

private node addwaiter(node mode) {
    //把当前线程封装成 node ，并且是独占模式
    node node = new node(thread.currentthread(), mode);
    //尝试快速入队，如果失败，则会调用 enq 入队方法。enq 会初始化队列。
    node pred = tail;
    //如果 tail 不为空，说明当前队列中已经有节点
    if (pred != null) { 
        //把当前 node 的 prev 指针指向 tail
        node.prev = pred;
        //通过 cas 把 node 设置为 tail，即添加到队尾
        if (compareandsettail(pred, node)) {
            //把旧的 tail 节点的 next 指针指向当前 node
            pred.next = node;
            return node;
        }
    }
    //当 tail 为空时，把 node 添加到队列，如果需要的话，先进行队列初始化
    enq(node);
    //入队成功之后，返回当前 node
    return node;
}

enq

通过自旋，把当前节点加入到队列中

private node enq(final node node) {
    for (;;) {
        node t = tail;
        //如果 tail为空，说明队列未初始化
        if (t == null) { 
            //创建一个空节点，通过 cas把它设置为头结点
            if (compareandsethead(new node()))
                //此时只有一个 head头节点，因此把 tail也指向它
                tail = head;
        } else {
            //第二次自旋时，tail不为空，于是把当前节点的 prev指向 tail节点
            node.prev = t;
            //通过 cas把 tail节点设置为当前 node节点
            if (compareandsettail(t, node)) {
                //把旧的 tail节点的 next指向当前 node
                t.next = node;
                return t;
            }
        }
    }
}

acquirequeued

入队成功之后，就会调用 acquirequeued 方法自旋抢锁。

final boolean acquirequeued(final node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            //获取当前节点的前驱节点
            final node p = node.predecessor();
            //如果前驱节点就是 head 节点，就调用 tryacquire 方法抢锁
            if (p == head && tryacquire(arg)) {
                //如果抢锁成功，就把当前 node 设置为头结点
                sethead(node);
                p.next = null; // help gc
                failed = false;
                //抢锁成功后，会把线程中断标志返回出去，终止for循环
                return interrupted;
            }
            //如果抢锁失败，就根据前驱节点的 waitstatus 状态判断是否需要把当前线程挂起
            if (shouldparkafterfailedacquire(p, node) &&
                //线程被挂起时，判断是否被中断过
                parkandcheckinterrupt())
                //注意此处，如果被线程被中断过，需要把中断标志重新设置一下
                interrupted = true;
        }
    } finally {
        if (failed)
            //如果抛出异常，则取消锁的获取，进行出队操作
            cancelacquire(node);
    }
}

sethead

通过代码，我们可以看到，当前的同步队列中，只有第二个节点才有资格抢锁。如果抢锁成功，则会把它设置为头结点。

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

需要注意的是，这个方法，会把头结点的线程设置为 null 。想一下，为什么？

因为，此时头结点的线程已经抢锁成功，需要出队了。自然的，队列中也就不应该存在这个线程了。

ps：由 enq 方法，还有 sethead 方法，我们可以发现，头结点的线程总是为 null。这是因为，头结点要么是刚初始化的空节点，要么是抢到锁的线程出队了。因此，我们也常常把头结点叫做虚拟节点（不存储任何线程）。

shouldparkafterfailedacquire

以上是抢锁成功的情况，那么抢锁失败了呢？这时，我们需要判断是否应该把当前线程挂起。

private static boolean shouldparkafterfailedacquire(node pred, node node) {
    //获取当前节点的前驱节点的 waitstatus
    int ws = pred.waitstatus;
    if (ws == node.signal)
        //如果 ws = -1 ，说明当前线程可以被前驱节点正常唤醒，于是就可以安全的 park了
        return true;
    if (ws > 0) {
        //如果 ws > 0，说明前驱节点被取消，则会从当前节点依次向前查找，
        //直到找到第一个没有被取消的节点，把那个节点的 next 指向当前 node
        //这一步，是为了找到一个可以把当前线程唤起的前驱节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitstatus > 0);
        pred.next = node;
    } else {
        //如果 ws 为 0，或者 -3（共享锁状态），则把它设置为 -1 
        //返回 false，下次自旋时，就会判断等于 -1，返回 true了
        compareandsetwaitstatus(pred, ws, node.signal);
    }
    return false;
}   

parkandcheckinterrupt

如果 shouldparkafterfailedacquire 返回 true，说明当前线程需要被挂起。因此，就执行此方法，同时检查线程是否被中断。

private final boolean parkandcheckinterrupt() {
    //把当前线程挂起，则 acquirequeued 方法的自旋就会暂停，等待前驱节点 unpark
    locksupport.park(this);
    //返回当前节点是否被中断的标志，注意此方法会把线程的中断标志清除。
    //因此，返回上一层方法时，需要设置 interrupted = true 把中断标志重新设置，以便上层代码可以处理中断
    return thread.interrupted();
}

想一下，为什么抢锁失败后，需要判断是否把线程挂起？

因为，如果抢不到锁，并且还不把线程挂起，acquirequeued 方法就会一直自旋下去，这样你的cpu能受得了吗。

cancelacquire

当不停的自旋抢锁时，若发生了异常，就会调用此方法，取消正在尝试获取锁的线程。node 的位置分为三种情况，见下面注释，

private void cancelacquire(node node) {

    if (node == null)
        return;

    // node 不再指向任何线程
    node.thread = null;

    node pred = node.prev;
    //从当前节点不断的向前查找，直到找到一个有效的前驱节点
    while (pred.waitstatus > 0)
        node.prev = pred = pred.prev;

    node prednext = pred.next;

    //把 node 的 ws 设置为 -1 
    node.waitstatus = node.cancelled;

    // 1.如果 node 是 tail，则把 tail 更新为 node，并把 pred.next 指向 null
    if (node == tail && compareandsettail(node, pred)) {
        compareandsetnext(pred, prednext, null);
    } else {
        int ws;
        //2.如果 node 既不是 tail，也不是 head 的后继节点，就把 node的前驱节点的 ws 设置为 -1
        //最后把 node 的前驱节点的 next 指向 node 的后继节点
        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 {
            //3.如果 node是 head 的后继节点，则直接唤醒 node 的后继节点。
            //这个也很好理解，因为 node 是队列中唯一有资格尝试获取锁的节点，
            //它放弃了资格，当然有义务把后继节点唤醒，以让后继节点尝试抢锁。
            unparksuccessor(node);
        }

        node.next = node; // help gc
    }
}

unparksuccessor

这个唤醒方法就比较简单了，

private void unparksuccessor(node node) {
    
    int ws = node.waitstatus;
    if (ws < 0)
        compareandsetwaitstatus(node, ws, 0);

    node s = node.next;
    if (s == null || s.waitstatus > 0) {
        s = null;
        //从尾结点向前依次遍历，直到找到距离当前 node 最近的一个有效节点
        for (node t = tail; t != null && t != node; t = t.prev)
            if (t.waitstatus <= 0)
                s = t;
    }
    if (s != null)
        //把这个有效节点的线程唤醒，
        //唤醒之后，当前线程就可以继续自旋抢锁了，（回到 park 的地方）
        locksupport.unpark(s.thread);
}

下面画一个流程图更直观的查看整个获取锁的过程。

公平锁

公平锁和非公平锁的整体流程大致相同，只是在抢锁之前先判断一下是否已经有人排在前面，如果有的话，就不执行抢锁。我们通过源码追踪到 fairsync.tryacquire 方法。会发现，多了一个 hasqueuedpredecessors 方法。

hasqueuedpredecessors

这个方法判断逻辑稍微有点复杂，有多种情况。

public final boolean hasqueuedpredecessors() {
    node t = tail; 
    node h = head;
    node s;
    return h != t &&
            ((s = h.next) == null || s.thread != thread.currentthread());
}

1. 如果 h == t，说明 h 和 t 都为空（此时队列还未初始化）或者它们是同一个节点（说明队列已经初始化，并且只有一个节点，此时为 enq 方法第一次自旋成功后）。此时，返回false。
2. 如果 h != t，则判断 head.next == null 是否成立，如果成立，则返回 true。这种情况发生在有其他线程第一次入队时。在 aqs 的 enq 入队方法，设置头结点成功之后 compareandsethead(new node()) ，还未执行 tail = head 时（仔细想一想为什么？）。此时 tail = null , head = new node()，head.next = null。
3. 如果 h != t，并且 head.next != null，说明此时队列中至少已经有两个节点，则判断 head.next 是否是当前线程。如果是，返回 false（注意是 false哦，因为用了 !），否则返回 true 。

总结：以上几种情况，只有最终返回 false 时，才会继续往下执行。因为 false，说明没有线程排在当前线程前面，于是通过 cas 尝试把 state 值设置为 1。若成功，则方法返回。若失败，同样需要去排队。

公平锁和非公平锁区别

举个例子来对比公平锁和非公平锁。比如，现在到饭点了，大家都到食堂打饭。把队列中的节点比作排队打饭的人，每个打饭窗口都有一个管理员，只有排队的人从管理员手中抢到锁，才有资格打饭。打饭的过程就是线程执行的过程。

如果，你发现前面没有人在排队，那么就可以直接从管理员手中拿到锁，然后打饭。对于公平锁来说，如果你前面有人在打饭，那么你就要排队到他后面（图中b），等他打完之后，把锁还给管理员。那么，你就可以从管理员手中拿到锁，然后打饭了。后面的人依次排队。这就是fifo先进先出的队列模型。

对于非公平锁来说，如果你是图中的 b，当 a 把锁还给管理员后，有可能有另外一个 d 插队过来直接把锁抢走。那么，他就可以打饭，你只能继续等待了。

所以，可以看出来。公平锁是严格按照排队的顺序来的，先来后到嘛，你来的早，就可以早点获取锁。优点是，这样不会造成某个线程等待时间过长，因为大家都是中规中矩的在排队。而缺点呢，就是会频繁的唤起线程，增加 cpu的开销。

非公平锁的优点是吞吐量大，因为有可能正好锁可用，然后线程来了，直接抢到锁了，不用排队了，这样也减少了 cpu 唤醒排队线程的开销。 但是，缺点也很明显，你说我排队排了好长时间了，终于轮到我打饭了，凭什么其他人刚过来就插到我前面，比我还先打到饭，也太不公平了吧，后边一大堆排队的人更是怨声载道。这要是每个人来了都插到我前面去，我岂不是要饿死了。

独占锁的释放

我们从 reentrantlock 的 unlock 方法看起：

public void unlock() {
    //调用 aqs 的 release 方法
    sync.release(1);
}

public final boolean release(int arg) {
    if (tryrelease(arg)) {
        node h = head;
        //如果头结点不为空，并且 ws 不为 0，则唤起后继节点
        if (h != null && h.waitstatus != 0)
            unparksuccessor(h);
        return true;
    }
    return false;
}

这段逻辑比较简单，当线程释放锁之后，就会唤醒后继节点。 unparksuccessor 已讲，不再赘述。然后看下 tryrelease 方法，公平锁和非公平锁走的是同一个方法。

protected final boolean tryrelease(int releases) {
    //每释放一次锁，state 值就会减 1，因为之前可能有锁的重入
    int c = getstate() - releases;
    //如果当前线程不是抢到锁的线程，则抛出异常
    if (thread.currentthread() != getexclusiveownerthread())
        throw new illegalmonitorstateexception();
    boolean free = false;
    if (c == 0) {
        //只有 state 的值减到 0 的时候，才会全部释放锁
        free = true;
        setexclusiveownerthread(null);
    }
    setstate(c);
    return free;
}

因为，reentrantlock 支持锁的重入，所以每次重入 state 值都会加 1，相应的每次释放锁， state 的值也会减 1 。所以，这也是为什么每个 lock 方法最后都要有一个 unlock 方法释放锁，它们的个数需要保证相同。

当 state 值为 0 的时候，说明锁完全释放。其他线程才可以有机会抢到锁。

结语

以上已经讲解了独占锁主要的获取方法 acquire ，另外还有一些其他相关方法，不再赘述，因为主要逻辑都是一样的，只有部分稍有不同，只要理解了 acquire ，这些都是相通的。如 acquireinterruptibly 方法，它可以在获取锁的时候响应中断。还有超时获取锁的方法 doacquirenanos 可以设定获取锁的超时时间，超时之后就返回失败。

下篇预告：分析 reentrantreadwritelock 读写锁源码，以及 aqs 共享锁的获取和释放，敬请期待。

如果本文对你有用，欢迎点赞，评论，转发。

学习是枯燥的，也是有趣的。我是「烟雨星空」，欢迎关注，可第一时间接收文章推送。