ReentrantLock原理学习-白红宇

ReentrantLock原理学习

阅读量：5155 次

发布时间：2019-06-13

本文共 11962 字，大约阅读时间需要 39 分钟。

　　我们学习了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是直接进入正常的锁获取流程，代码如下所示；

// 此为FairLockfinal void lock() {
    　　 // 一上来就尝试获取锁    if (compareAndSetState(0, 1))        setExclusiveOwnerThread(Thread.currentThread());    else        acquire(1);}// 此为NonfairLockfinal 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;}