Android技术知识Android开发经验谈Android葵花宝典

ReentrantLock源码详解

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

本文主要内容

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则是在正常等待状态。

添加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,被虐后居然还有一丝快感。。

上一篇下一篇

猜你喜欢

热点阅读