ReentrantLock原理学习

　　上文我们学习了ReentrantLock的基本用法，在最后我们留下了一个问题，ReentrantLock获取的锁是什么锁呢？本文我们就从源码的角度来一探究竟。本文涉及到的源码对应JDK版本为1.8。

　　上文说到，ReentrantLock常用的获取锁方式为：

ReentrantLock lock = new ReentrantLock();
lock.lock();
try {
    // do something
} finally {
     lock.unlock();
}

　　好，我们就从这个地方开始，首先构造一个ReentrantLock对象，我们先来看一下构造器做了什么：

构造ReentrantLock实例

private final Sync sync;

// 初始化，构造一个非公平同步锁
public ReentrantLock() {
    sync = new NonfairSync();
}

abstract static class Sync extends AbstractQueuedSynchronizer {
　　
　　abstract void lock();

　　...
}
// 非公平锁
static final class NonfairSync extends Sync {
　　...
}
// 公平锁
static final class FairSync extends Sync {
　　...
}

　　可以看到在构造器中创建了一个NonfairSync对象，这个NonfairSync是在ReentrantLock内部定义的一个静态类，继承自ReentrantLock内部另一个抽象内部类Sync(其中定义了用于获取锁的抽象方法lock()，NonfairSync实现了该方法用于获取非公平锁，同时还有一个FairSync也实现了该方法，用于获取公平锁)，这个Sync是继承自AbstractQueueSynchronizer，简称AQS。

---

 　　在开发中的大多数情况下我们都是不会直接使用AQS的，因为标准同步器类(比如ReentrantLock、CountDownLatch)的集合能够满足绝大多数情况的需求。但是如果能了解标准同步器类的实现方式，那么对于理解它们的工作原理是非常有帮助的。

　　在基于AQS构建的同步器类中，最基本的操作包括各种形式的获取操作和释放操作。获取操作是一种依赖状态的操作，并且通常会阻塞。当使用锁或信号量时，“获取”操作的含义很直观，即获取的是锁或者许可，并且调用者可能会一直等待直到同步器类处于可被获取的状态。AQS负责管理一个整数状态(state)信息，可以通过getState,setState以及compareAndSetState等protected类型方法来进行操作。这个整数可以用于表示任意状态。在ReentrantLock中用它来表示所有者线程已经重复获取该锁的次数。AQS还可以管理一些额外的状态变量，在ReentrantLock中可以保存锁的当前所有者(线程)的信息，这样就能区分某个获取操作是重入的还是竞争的，这是通过setExclusiveOwnerThread实现的。

---

1. 获取锁

　　通过lock()来获取锁(这里是获取的非公平锁)：

// 获取锁
public void lock() {
    sync.lock();
}

　　调用sync的lock()方法，从上面的构造函数我们可以知道这里的sync实际对象类型是NonfairSync，顾名思义，是非公平锁，所以。接下来看看NonfairSync中对lock()的实现：

/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/

final void lock() {

　　// 一进来就利用CAS机制尝试修改state

　　if (compareAndSetState(0, 1))
　　　　setExclusiveOwnerThread(Thread.currentThread());　　　　// 将当前线程置为独占者
　　else
　　　　acquire(1);
}

// AQS中的方法，获取锁

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

1.1 尝试进入独占模式

　　可以看到，在NonfairSync的lock()方法中，一上来就直接调用AQS中的compareAndState()方法去尝试修改state变量(通过CAS机制)，如果修改成功则将当前线程置为获得锁的独占线程(setExclusiveOwnerThread()方法)。如果修改状态失败则调用AQS中的acquire()方法。

　　在acquire()方法中会先调用在子类(NonfairSync)中实现的tryAcquire()方法来尝试获取锁，tryAcquire的代码如下：

// 该方法在NonfairSync中有实现，直接调用其父类Sync中的方法nonfairTryAcquire()
protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}

// 实现在Sync中
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {　　// 状态量state为0代表锁未被抢占，则可以尝试获取锁
        if (compareAndSetState(0, acquires)) {　　// 获取锁的操作为将状态量state加上一个值(传入的acquires，一般为1)，成功后将当前线程置为独占者，返回true代表获取锁成功
            setExclusiveOwnerThread(current);
            return true;
        }
    }
　　 // 如果state不为0，则判断当前线程是否为独占线程，是则将state加上acquire，这是可重入锁的体现
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

　　tryAcquire()方法调用了nonfairTryAcquire，其实现是在Sync中，主要做了以下几件事情：

• 尝试修改state，如果修改成功则将当前线程置为获得锁的独占线程；
• 如果修改state失败，则判断当前线程是否是已经获得锁的独占线程，如果是则将state加1(为什么是1，需要结合上下文理解，因为在lock方法中，acquire方法传入的就是1)；
• 如果不是则返回false，代表获取锁失败；

　　这里有个细节，就是明明前面才尝试过获取锁(就是刚进入lock()方法时尝试修改state)，而接下来调用tryAcquire时又尝试获取锁，我的理解，类似synchronized自旋的原理，在将其加入等待队列之前再尝试获取一次，成功则直接获取锁，失败则加入等待队列中。

1.2 加入阻塞队列

　　接着，如果tryAcquire获取锁失败则会执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)，这里会先执行addWaiter(Node.EXCLUSIVE)，其实现是在AQS中：

// 因为未抢到锁，所以将线程放到等待队列中等待
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;
        }
    }
　　 // 如果tail为null，则进入enq方法
    enq(node);
    return node;
}

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
　　　　 // tail为null，则利用cas机制尝试new一个Node更新到tail中
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
　　　　　　　// 从队尾入队
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
} 

　　先看一下Node，这里的Node是实现在AQS内部的一个数据结构，作用是用来保存未获取到锁的线程(内部数据结构是一个双向链表，实现一个等待队列用于阻塞同步器)，每一个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;
　　// 前一个node
    volatile Node prev;
　　// 下一个node
    volatile Node next;
　　// 代表竞争锁的线程
    volatile Thread thread;

    Node nextWaiter;

    。。。
}

　　其中的waitStatus代表状态位，有五种取值，分别为：

• CANCELLED(1)，处于取消状态的节点是由于超时或者中断导致，该节点其状态不会再改变，并且该节点代表的线程不会再阻塞；
• SIGNAL(-1)，当前节点的继任者已经或者即将阻塞(通过park)，所以当当前节点释放或者取消时必须unpark其继任者。为避免竞争，在acquire方法中必须先指示需要signal信号，然后重试原子的acquire方法，之后如果失败则阻塞；
• CONDITION(-2)，该状态表示该节点当前处于等待队列中，而不会作为一个同步队列的节点，直到节点的状态值变成0(这里0并不像其他状态一样，是不代表意义的，仅仅只是起简化作用)；
• PROPAGATE(-3)，共享模式下会用到；
• 0，除了如上四种状态，新new出来的node是这个值； 

　　

　　好了，了解了Node之后我们再来看上面的代码，在addWaiter方法中：

• 首先new了一个Node对象，参数是Node.EXCLUSIVE；
• 获取队尾node(tail)，判断是否存在；
• 如果存在，则直接将当前线程(第一步new的node)替换tail，完成入队；
• 如果不存在，则调用enq()方法；

　　enq()方法逻辑如下：

• 取队尾元素tail；
• 如果为空，则利用cas将一个新的node设置为对头head，如果成功则将tail也指向head；
• 如果不为空，则直接将传入的node置为队尾；

　　至此，将当前线程加入等待队列的操作就完成了。

1.3 将当前线程阻塞或继续执行 

　　接下来，执行acquireQueued(Node node,int arg)方法，主要是在将未获取到锁的当前线程加入到等待队列之后，再将当前线程阻塞或者自旋。因为该方法中有自旋操作，所以线程进入方法之后要么阻塞等待唤醒，要么被唤醒，并且这时已经进入等待队列了，该方法结束时代表线程已经去阻塞队列中转了一圈，被唤醒了，所以叫acquireQueued，意味着从Queue中获取一个node：

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) { // 自旋操作
            final Node p = node.predecessor();　　// 获取该节点的前驱节点
　　　　　　　// 如果当前节点的前驱节点是等待队列的头结点且当前节点tryAquire成功(修改状态成功)，则将当前节点置为头节点并且返回false
            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);
    }
}

// 判断是否需要阻塞
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.
         */
        return true;
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        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.
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);　　　　　　// 线程阻塞
    return Thread.interrupted();　　 // 中断
}

 　　acquireQueued方法的主要逻辑为：

• 获取当前节点的前驱节点；
• 如果当前节点的前驱节点是等待队列的头结点且当前节点tryAquire操作成功(修改状态成功，也就是成功获取锁)，则直接将当前节点置为头节点并且返回false；
• 如果当前节点的前驱节点不是头结点或者tryAcquire操作失败，则会执行shouldParkAfterFailAcquire方法和parkAndCheckInturrupt方法；
• 如果shouldParkAfterFailAcquire方法和parkAndCheckInturrupt方法都返回true，则会将interrupted置为true，然后返回第1步操作，继续循环(也是自旋)；

　　shouldParkAfterFailAcquire方法主要逻辑是判断是否需要将当前线程阻塞：

• 获取当前节点前驱节点的waitSatus，后面简写为ws；
• 如果ws为Node.SIGNAL，则直接返回true；
• 如果ws大于0，即为Node.CANCELLED，则循环将当前节点前驱节点的前驱节点置为当前节点的前驱节点，直到前驱节点的ws不为Node.CANCELLED；
• 如果ws小于等于0且不为Node.SIGNAL，此时可能的值为0(代表新node)或-3(Node.PROPAGATE)，则尝试将ws修改为Node.SIGNAL；
• 返回false，进入下一次自旋；

　　parkAndCheckInturrupt方法主要作用则是将当前线程阻塞且检查中断状态：

• 利用LockSupport.park()方法将当前线程阻塞；
• 解除阻塞之后，返回当前线程是否被打断；

　　至此，锁获取操作已经完成，配合下面的时序图可以更好理解整个过程：

3. 释放锁　　

　　下面我们再来看锁的释放操作。

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

　　可以看到在unlock()方法中是直接调用sync的父类AbstractQueuedSynchronized中的release()方法，在该方法中：

• 先尝试tryRelease，返回false则直接返回false，锁未释放；
• 如果tryRelease返回true则将唤醒head节点的继任者；

3.1 尝试释放锁

　　tryRelease()方法是定义在AQS中的方法，需要在子类中实现，我们这里看其在Sync中的实现：

    protected final boolean tryRelease(int releases) {
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }

　　tryRelease中主要做了以下几件事情：

• 将state减去1(这里传入的release为1)；
• 如果当前线程不为独占线程则抛出异常；
• 如果状态值state已经减为0则将独占线程置为空，state置为0，返回true，代表release成功；
• 否则将state减去传入的release值之后更新，并且返回false，代表未释放锁；

3.2 唤醒head节点的继任者

　　tryRelease成功代表着修改状态量state成功了，之后还需要唤醒head节点的继任者，让其去竞争锁，主要逻辑在unParkSuccessor()方法中：

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

 　　这里的逻辑主要是：

• 如果头节点的waitStatus为负，则将其置为0，失败了也没关系；
• 需要唤醒的继任者一般是头节点的下一个节点，但是如果该节点为空或者该节点statu是cancelled，则需要循环从等待队列末尾往前选择合格继任者，直到队首；
• 如果有继任者，则通过LockSupport.unpark()将其唤醒，让其去竞争锁；

　　这里有没有想过，为什么此时是让继任者去竞争锁而不是直接将锁给它呢？

4. 公平锁与非公平锁的区别

　　这里就要涉及到公平锁和非公平锁的区别了，所谓公平，在这里是指等待时间越久获取锁的优先级越高，或者说等待时间久的线程更容易获取到锁。在ReentrantLock这里是如何实现的呢？

　　在获取锁时有两处不一样：

• 在开始lock时，NonfairLock是不管三七二十一，先直接去竞争一次锁，然后再进入acquire尝试正常的获取锁流程，而FairLock是直接进入正常的锁获取流程，代码如下所示；

// 此为FairLock
final void lock() {
　　 // 一上来就尝试获取锁
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

// 此为NonfairLock
final void lock() {
    acquire(1);
}

• 而进入正常获取锁流程之后，两者的tryAcquire方法实现略有不同，在检测到没有线程占用锁之后FairLock还会检查一遍它的前面是否有等待的节点，如果有则是不会获取锁的；而NonfairLock则没有这一步检查，直接尝试抢占锁，公平锁的代码如下：

protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        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;
}

5. 总结

　　至此，我们已经分析完了ReentrantLock完整的获取和释放锁过程，总结一下：

• ReentrantLock锁机制底层是依赖于AQS实现的，获取锁操作实质就是修改ReentrantLock对象的state状态量，修改成功的线程则获取锁并继续执行，并且该线程会成为锁的独占线程，失败的线程则会放到一个等待队列中并阻塞等待唤醒。这就是本文通过阅读源码得出的答案；
• 线程释放锁的时候会将状态量state递减，到0之后代表锁释放成功，则将独占线程清空，并且唤醒一个等待节点(如果有的话)；
• 公平锁和非公平锁的区别在于获取锁的时候公平锁会按照等待先后顺序获取，而非公平锁则是抢占式的获取锁。两者释放锁的流程是一样的；