[源码分析]ReentrantLock & AbstractQueuedSynchronizer

Posted 2020-12-04 noking

[源码分析]ReentrantLock & AbstractQueuedSynchronizer

首先声明一点: 我在分析源码的时候, 把jdk源码复制出来进行中文的注释, 有时还进行编译调试什么的, 为了避免和jdk原生的类混淆, 我在类前面加了"My". 比如把ReentrantLock改名为了MyReentrantLock, 在源码分析的章节里, 我基本不会对源码进行修改, 所以请忽视这个"My"即可.

---

一. sync字段

首先来看一下ReentrantLock里唯一的一个字段

Sync继承自AQS(AbstractQueuedSynchronizer, 以下简称AQS) . 公平锁和非公平锁都继承了Sync. Sync是ReentrantLock类里锁的统一声明. 

二. lock/unlock依赖Sync

ReentraintLock的 lock()和unlock()方法实际上都是靠Sync来实现的:

三. 锁内部类定义

Sync 和 公平锁 和 非公平锁 都是ReentrantLock的内部类, 类的定义部分如下(细节先隐藏起来了, 后面会讲):

四. ReentrantLock构造器

ReentrantLock有两个构造器.

1. 默认构造器是直接使用了非公平锁. 非公平锁就是不一定按照"先来后到"的顺序来进行争抢.

2. 带参构造器可以传递一个bool类型. true的时候为公平锁. 公平锁就是按照"先来后到"的顺序来进行争抢.

五. 公平锁获取锁的流程(单线程, 没有争抢) 

首先从最外层的调用lock()方法开始咱们在Main方法里写下这两行代码:

MyReentrantLock就是ReentrantLock, 我复制了源代码, 然后改了个名字而已.

Reentraint类的lock()方法最终还是调用的sync.lock()

由于我们现在使用的是公平锁. 所以sync现在是FairSync. 所以sync.lockI()实际上就是FairSync类里的lock()方法

发现lock()调用的是acquire(1)这个方法, 这个方法是在AQS类里实现的.代码如下:

arg当时传进来的是1, 所以首先进行的是tryAcquire(1)来进行"尝试获取锁"的操作. 这时一种乐观的想法.

tryAcquire方法的具体实现在FairSync类里, 具体代码如下:

/**
     * @return 返回true: 获取到锁; 返回false: 未获取到锁
     * 什么时候返回true呢?  1.没有线程在等待锁；2.重入锁，线程本来就持有锁，也就可以理所当然可以直接获取
     * @implNote 尝试直接获取锁.
     */
    protected final boolean tryAcquire(int acquires) {
        // 获取当前线程的引用
        final Thread current = Thread.currentThread();

        // 当前锁的计数器. 用于计算锁被获取的次数.在重入锁中表示锁重入的次数.由于这个锁是第一次被获取, 所以c==0
        int c = getState();

        // c==0, 也就是 state == 0 ,重入次数是0, 表示此时没有线程持有锁.
        if (c == 0) {
            // 公平锁, 所以要讲究先来后到
            // 因为有可能是上一个持有锁的线程刚刚释放锁, 队列里的线程还没来得及争抢, 本线程就乱入了
            // 所以每次公平锁抢锁之前, 都要判断一下等待队列里是否有其他线程
            if (!hasQueuedPredecessors() &&
                    // 执行到这里说明等待队列里没有其他线程在等待.
                    // 如果没有线程在等待，那就用CAS尝试一下，成功了就获取到锁了，
                    // 不成功的话，只能说明一个问题，就在刚刚几乎同一时刻有个线程抢先了 =_=
                    compareAndSetState(0, acquires)) {
                
                // 到这里就获取到锁了，标记一下，告诉大家，现在是我(当前线程)占用了锁
                setExclusiveOwnerThread(current);
                // 成功获取锁了, 所以返回true
                return true;
            }


            //-- 由于现在模拟的是单纯地获取一次锁, 没有重入和争抢的情况, 所以执行不到这里, 上面的cas肯定会成功, 然后返回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;
    }

争抢完锁之后会返回true, 然后回到上层方法acquire : 

if语句里 && 前面是false, 不会继续往下执行了. 当前线程获取到了锁, 而且执行了所有该执行的内容, 就完事儿了.

六. 公平锁进行重入的流程

重入就是一个线程获取到了锁, 然后这个线程又一次申请(进入)了这个锁.

重入用synchronized来举例就是这样:

用ReentrantLock来举例子就是这样:

同一个线程(main线程) 首先进行了lock.lock()申请并占有了锁, 随后又执行了一次lock.lock(). 还没释放锁的情况下, 又一次申请锁. 这样就是重入了.  

上面一小节已经分析了第一行的lock.lock()是如何获取到锁的, 所以我们只分析 重入的部分, 也就是后面那句lock.lock()的执行流程.

前面的执行过程一直是一模一样的, 直到这里:

/**
     * @return 返回true: 获取到锁; 返回false: 未获取到锁
     * 什么时候返回true呢?  1.没有线程在等待锁；2.重入锁，线程本来就持有锁，也就可以理所当然可以直接获取
     * @implNote 尝试直接获取锁.
     */
    protected final boolean tryAcquire(int acquires) {
        // 获取当前线程的引用
        final Thread current = Thread.currentThread();

        // 当前锁的计数器. 由于前面的那句lock已经获取到锁了, 所以这里是status==1, 也就是 c==1
        int c = getState();

        // c==1, 表示当前有线程持有锁, 所以这段if是进不去了
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }

        // 由于 c==1 , 无法进入if语句, 所以来看看满不满足这里的 else if
        // 这个锁被人占了, 但还是不死心, 于是看一下是不是当前线程自己占的这个锁.
        // (人家女生说有喜欢的人, 为什么不问问是不是自己呢 = =.)
        // 由于是同一个线程, 所以就是自己啦! 所以会进入这个else if分支，
        } else if (current == getExclusiveOwnerThread()) {
            // 代码执行到这里了, 就是所谓的 重入 了

            // 这里的acquires的值是1, 所以nextc =  1 + 1 , 也就是2了
            int nextc = c + acquires;
            // 小于0, 说明int溢出了
            if (nextc < 0) throw new Error("Maximum lock count exceeded");
            // 在这里把状态更新一下, 把state更新为2, 意思就是这个锁被同一个线程获得2次了. 
            // (大家就可以以此类推, 下次再重入的话, 那么就会再+1, 就会变为3....)
            setState(nextc);
            // 重入完成, 返回true
            return true;
        }
        
        return false;
    }

 还记得上小节讲的, 获取锁的时候进入的是这段代码的if语句, 而重入就不一样了, 进入的是 else if语句. 但最终返回的还是true, 表示成功. 

上面讲的是无争强的情况, 接下来讲讲有争抢的情况.

cas争抢失败

场景如下:

一开始锁是空闲状态, 然后两个线程同时争抢这把锁(在cas操作处发生了争抢).

一个线程cas操作成功, 抢到了锁; 另一个线程cas失败. 

代码例子如下(代码的意思到位了, 但是这段代码最后不一定会在cas处进行争抢, 大家意会就好了):

cas操作成功的线程就和第五小节的一样, 就不用再重复描述了.

而cas争抢失败的线程会何去何从呢? 看我给大家分析: 

 /**
     * @return 返回true: 获取到锁; 返回false: 未获取到锁
     * 什么时候返回true呢?  1.没有线程在等待锁；2.重入锁，线程本来就持有锁，也就可以理所当然可以直接获取
     * @implNote 尝试直接获取锁.
     */
    protected final boolean tryAcquire(int acquires) {
        // 获取当前线程的引用
        final Thread current = Thread.currentThread();

        // 当前锁的计数器.
        int c = getState();

        // state == 0 表示此时没有线程持有锁
        if (c == 0) {
            // 本场景中, 一开始锁是空闲的, 所以队列里没有等待的线程
            if (!hasQueuedPredecessors() &&
                    // 两个线程在这里进行争抢
                    // cas抢成功的会进入到if代码块
                    // cas抢失败的, 就跳出整个if-else, 也就是直接到最后一行代码
                    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;
        }

        // cas 操作失败后, 会这直接执行到这里. 返回false.
        return false;
    }

 在这里返回了false, 回到上一层函数.

第一个条件是true, 所以会继续往下执行acquireQueued方法. 来准备让这个失败的线程进入队列等待.

下面继续来给大家讲解 acquireQueued(addWaiter(Node.EXCLUSIVE), arg) .

先讲讲这个addWaiter(Node.EXCLUSIVE):

/**
     * 将当前线程封装为Node, 然后根据所给的模式, 进行入队操作
     *
     * @param mode 有两种模式 Node.EXCLUSIVE 独占模式, Node.SHARED 共享模式
     * @return 返回新节点, 这个新节点封装了当前线程.
     */
    private Node addWaiter(Node mode) { // 这个mode没用上.
        Node node = new Node(Thread.currentThread(), mode);
        // 咱们刚才都没见到过tail被赋予了其他的值, 当然就是null了.
        Node pred = tail;
        // tail是null的话, pred就是null, 所以不会进入到这个if语句中.所以跳过这个if语句.
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }

        // 因为锁的等待队列是懒初始化, 直到有节点插入进来, 它才初始化.
        // 而现在这个挣钱失败的线程, 正好是锁建立以来, 第一个进入等待队列的线程. 所以现在才准备进行初始化.
        // 初始化完了后会把当前线程的相关信息和引用封装成Node节点, 然后插入到队列当中.并且制定head 和 tail.
        // tail就不等于null了, 所以下一次addWaiter方法被调用的时候, 就会执行上面的if语句了. 而不会跳过if语句, 来到这里进行初始化了.
        enq(node);
        // 返回这个Node节点.
        return node;
    }

 目的就是要将这个cas失败的线程封装成节点, 然后插入到队尾中. (等待队列是懒初始化,) 

如果队列已经初始化了, 那么tail就不会是null, 就会执行上面代码中的if语句, 调整一下指针的引用就好了.

但是如果队列还未初始化, 那么就应该先初始化, 再插入. 先初始化,再插入, 对应的代码是enq(node). 

接下来讲解一下enq方法: 

   /**
     * 采用自旋的方式入队
     * CAS设置tail,直到争抢成功.
     */
    private Node enq(final Node node) {
        for (; ; ) {
            Node t = tail;
            //  最开始tail肯定是null, 进入if进行初始化head和tail.
            if (t == null) { // Must initialize
                // 设置head 和tail. cas来防止并发.
                if (compareAndSetHead(new Node())) tail = head;
                
            // if 语句执行完了后, 之后的for循环就会走else了.
            } else {
                // 争抢入队, 没抢到就继续for循环迭代.抢成功了就可以return了,不然一直循环.
                // 为什么是用cas来争抢呢? 因为怕是多个线程一起执行到这里啊 
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

　　

未完待续