ReentrantLock源码详解

2018-08-12 本文已影响7人某昆

本文主要内容

ReentrantLock简要介绍
lock流程
unlock流程
总结

1、ReentrantLock简要介绍

加锁可以使用 synchronized 关键字，也可以使用 ReentrantLock 对象。synchronized 加锁后，会在同步代码块内添加 monitorenter他monitorexit这两个字节码指令。而 ReentrantLock 加锁，则是api层面实现的，它主要依赖于 Unsafe类的线程挂起和恢复功能。

二者之间的区别可以参考线程安全一文，本文中着重阐述下 ReentrantLock 加锁与释放锁的流程。

private final Sync sync;

ReentrantLock内部有一个成员变量，sync，它继承自 AbstractQueuedSynchronizer类，同时 sync 有两个实现类，NonfairSync 和 FairSync类，分别代表着非公平锁模式和公平锁模式。ReentrantLock内部的 sync 成员变量默认是非公平锁模式。

其实 ReentrantLock 的逻辑有一大半依靠 AbstractQueuedSynchronizer 类实现，它就是大名顶顶的 AQS，AQS定义了一套多线程访问共享资源的同步器框架，许多同步类实现都依赖于它，如常用的ReentrantLock/Semaphore/CountDownLatch...

顾名思义，AQS，抽象队列同步器，它的内部确实有一个链表队列，队列中为排队等待锁的Node。

它维护了一个volatile int state（代表共享资源）和一个FIFO线程等待队列（多线程争用资源被阻塞时会进入此队列）。当state值为1时，则表示当前已经有线程取得了锁，其它线程则需要被阻塞。反之，如果state值为0，则表示锁还没有被任何一个线程获取。

AQS定义两种资源共享方式：Exclusive（独占，只有一个线程能执行，如ReentrantLock）和Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）。本文中只讲述 ReentrantLock 类，故只阐述独占模式。

2、lock流程

先看看 lock 方法，由于默认实现是非公平锁，故本文也只会讲述非公平锁的逻辑实现。

final void lock() {
        //先尝试将state值由0设置为1，并且将自己设置为独占线程
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            //如果state值设置失败，则调用acquire方法，这是核心方法
            acquire(1);
    }

lock 方法，刚开始时就尝试将state值由0变1，如果当前的state值为0，表明没有线程获取了锁，此时，如果lock方法将state值设置成功，则表明成功抢占锁。这也表明非公平锁的操作模式，直接抢占，不管公平与否。

如果state值变化失败，锁抢占失败，则调用 acquire 方法，acquire方法是在AQS类中。这也是加锁的关键方法。

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

整个方法看似挺简单的，首先调用 tryAcquire 方法。tryAcquire在AQS中是一个抽象方法，所以它的实现是在非公平锁当中。

protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        //如果c等于0，则尝试将state值置为1，抢占锁并设置自己为独占线程
        if (c == 0) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        //如果当前的线程等于独占线程，则表示锁重入了，则c加1
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        //返回false，则表明抢占锁失败
        return false;
    }

非常重要的一点，ReentrantLock它是可重入锁，state的值可能大于1。从代码可知，ReentrantLock 加锁时，lock了多少次，则一定要unlock多少次，这俩是配套的，如果少unlock了，则state的值不为0，则锁永远也释放不掉。

回到 acquire 方法，如果tryAcquire返回为false，则表示抢占锁失败，则会执行 addWaiter、acquireQueued 方法。

private Node addWaiter(Node mode) {
    //构造node对象，并将当前线程的引用传递过去
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    //将node置在链表尾端，先进先出的队列
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    //enq方法其实也是一种保险措施，保证node在链表尾端，自旋操作
    enq(node);
    return node;
}

值得注意一下，Node的状态一共有5种，接下来会用到，一般来说，如果状态大于0，处于cancel状态的Node，表明该节点已经取消等待了，需要被移除出等待队列了。少于0的Node则是在正常等待状态。

CANCELLED：值为1，在同步队列中等待的线程等待超时或被中断，需要从同步队列中取消该Node的结点，其结点的waitStatus为CANCELLED，即结束状态，进入该状态后的结点将不会再变化。
SIGNAL：值为-1，被标识为该等待唤醒状态的后继结点，当其前继结点的线程释放了同步锁或被取消，将会通知该后继结点的线程执行。说白了，就是处于唤醒状态，只要前继结点释放锁，就会通知标识为SIGNAL状态的后继结点的线程执行。
CONDITION：值为-2，与Condition相关，该标识的结点处于等待队列中，结点的线程等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
PROPAGATE：值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。
0状态：值为0，代表初始化状态。

添加Node到队尾线束后，继续查看 acquireQueued 方法

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        //注意，它是死循环
        for (;;) {
            //p为前节点
            final Node p = node.predecessor();
            //如果前节点已经是haed节点了，此时可以尝试获取下锁，
            //如果获取锁成功了，就将自己设置为head节点并且回收原head节点。
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //设置前节点的状态为 SIGNAL，并且挂起当前线程
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        //如果某种情况下，跳出了死循环，并且failed值为true，则需要删除当前节点，取消等待涣
        if (failed)
            cancelAcquire(node);
    }
}

从前文对 SIGNAL 状态的介绍可知，一定要将前节点状态设置为SIGNAL，shouldParkAfterFailedAcquire方法就是确保前节点的状态的。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    //如果前节点状态为 SIGNAL ，则直接返回 true退出。
    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) {
        //如果前节点状态大于0，即是 CANCELLED 状态，
        //这种状态的节点已经不等待了，它的节点也是非法节点了，所以需要路过
        /*
         * 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.
         */
        //通过CAS将前节点状态设置为 SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

将前节点状态设置完成后，需要将自身线程挂起。

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

昨天介绍的Unsafe类出场了，本文中还有好多CAS操作的，有兴趣的朋友可以查看下说说Java的Unsafe类一文。LockSupport类正是调用了Unsafe类的线程挂起方法，实现了线程阻塞。

到目前为止，lock方法已经阐述完毕了。接下来我们看看 unlock方法

3、unlock流程

先看看代码，unLock方法其实比较简单

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

先看看 tryRelease 的实现，按照之前 tryLock的套路，这个方法同样也是在子类中实现的。

protected final boolean tryRelease(int releases) {
        //c的值仅仅是减去传进来的参数1，这也说明加锁和释放锁一定要配套
        int c = getState() - releases;
        //如果当前线程并不等于独占线程，则报错，都没获得锁释放啥东西呢
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        //如果c等于0则表示锁真正释放了，将独占线程设置为null
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        //设置c的值
        setState(c);
        return free;
    }

回到release方法，如果tryRelease成功释放锁并且返回为true的话，则调用 unparkSuccessor 方法，这个方法应该就是唤醒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;
    //先将head节点的状态设置为0，也就是默认值
    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.
     */
    //寻找head节点的下一个处于正常等待状态的节点
    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);
}

整个方法果然和设想中的一致，将头节点的下个节点线程恢复，下个线程可以去竞争锁啦。仔细回想一下，貌似并没有地方让下个线程去竞争锁啊。回到之前的 acquireQueued 方法，里边是死循环的，所以其实重新竞争锁是发生在死循环中的，并不需要再单独去调用某方法了。

4、总结

ReentrantLock的代码还是非常的有意思的，之前我老在想竞争的锁到底是啥？线程是如何恢复和挂起的？仔细的回想这些细节问题，一步步推敲，到网上也找找资料，看明白它的源码就非常简单了。read the fucking source code，被虐后居然还有一丝快感。。

ReentrantLock源码详解

猜你喜欢

热点阅读