《吊打面试官》从根上剖析ReentrantLock的来龙去脉

一时失言乱红尘 2023-09-25 22:57 10阅读 0赞

# ReentrantLock #

## 一、引言 ##

**并发编程**在互联网技术存储方面使用如此广泛，几乎所有的后端技术面试官都要在**并发编程**的使用和原理方面对小伙伴们进行 360° 的刁难。

作为一个在互联网公司面一次拿一次 `Offer` 的面霸，打败了无数竞争对手，每次都只能看到无数落寞的身影失望的离开，略感愧疚（**请允许我使用一下夸张的修辞手法**）。

于是在一个寂寞难耐的夜晚，**暖男**我痛定思痛，决定开始写\*\*《吊打面试官》\*\*系列，希望能帮助各位读者以后面试势如破竹，对面试官进行 360° 的反击，吊打问你的面试官，让一同面试的同僚瞠目结舌，疯狂收割大厂 `Offer`！

**虽然现在是互联网寒冬，但乾坤未定，你我皆是黑马！**

![format_png][]

## 二、使用 ##

public class ReentrantLockTest {
        private static int count;
    
        public static void main(String[] args) {
    
            ReentrantLock lock = new ReentrantLock();
            Thread thread1 = new Thread(() -> {
                // 加锁
                lock.lock();
                for (int i = 0; i < 10; i++) {
                    System.out.println(Thread.currentThread().getName() + " : " + count++);
                }
                // 解锁
                lock.unlock();
    
            });
            thread1.setName("Thread-1");
            thread1.start();
    
    
            Thread thread2 = new Thread(() -> {
                // 加锁
                lock.lock();
                for (int i = 0; i < 10; i++) {
                    System.out.println(Thread.currentThread().getName() + " : " + count--);
                }
                // 解锁
                lock.unlock();
    
            });
            thread2.setName("Thread-2");
            thread2.start();
    
    
        }
    }
    
    复制代码

最终输出：

Thread-1 : 0
    Thread-1 : 1
    Thread-1 : 2
    Thread-1 : 3
    Thread-1 : 4
    Thread-1 : 5
    Thread-1 : 6
    Thread-1 : 7
    Thread-1 : 8
    Thread-1 : 9
    Thread-2 : 10
    Thread-2 : 9
    Thread-2 : 8
    Thread-2 : 7
    Thread-2 : 6
    Thread-2 : 5
    Thread-2 : 4
    Thread-2 : 3
    Thread-2 : 2
    Thread-2 : 1
    复制代码

通过 `ReentrantLock` 锁让我们的线程安全的交替打印 `count` 的数值，那这个到底是如何实现的呢？

## 三、源码 ##

从 `ReentrantLock` 源码来说，主要的方法如下：

*  加锁：lock、tryLock、lockInterruptibly
 *  释放锁：unlock

### 1、ReentrantLock 结构 ###

我们先来看一下整体如下所示：

![format_png 1][]

我们简单的介绍一下每部分的结构：

// 头结点
    private transient volatile Node head;
    
    // 尾结点
    private transient volatile Node tail;
    
    // 状态标记(判断当前线程是否拿到锁的标记)
    private volatile int state;
    
    // 当前占领锁的线程
    private transient Thread exclusiveOwnerThread;
    复制代码

我们看下 `Node` 节点的数据：

class Node {
        // 共享锁、互斥锁
        static final Node SHARED = new Node();
        static final Node EXCLUSIVE = null;
        
        // 当前节点的等待状态，默认为0
        volatile int waitStatus;
        // 表示线程获取锁的请求已经取消了
        static final int CANCELLED =  1;
        // 表示线程已经准备好了，就等资源释放了
        static final int SIGNAL    = -1;
        // 表示节点在等待队列中，节点线程等待唤醒
        static final int CONDITION = -2;
        // 当前线程处在SHARED情况下，该字段才会使用
        static final int PROPAGATE = -3;
    
        // 指向前继节点的指针
        volatile Node prev;
    	
        // 指向后继节点的指针
        volatile Node next;
    
        // 当前Node节点持有的线程
        volatile Thread thread;
    
        // 暂时可忽略
        Node nextWaiter;
    }
    复制代码

我们简单的过一下上面的结构，具体的使用我们后面源码里面会进行讨论。

### 2、Lock 源码 ###

对于 `ReentrantLock` 里面的加锁源码，一共有三种类型：

*  lock
 *  tryLock
 *  lockInterruptibly

我们挨个进行解读

2.1 lock()

public void lock() {
        // 这里的lock实现一共有两种
        // 公平锁与非公平锁
        sync.lock();
    }
    复制代码

我们先来看非公平锁的代码：

final void lock() {
        // 尝试CAS将当前的State从0置为1
        // 这里直接CAS也符合我们非公平锁的性格，不管你三七二十一，我直接CAS尝试一下能不能拿到锁
        if (compareAndSetState(0, 1))
            // 如果真让这哥们给拿到了，那没办法，就只能将AQS里面的占领锁的线程给设置成当前请求的线程
            setExclusiveOwnerThread(Thread.currentThread());
        else
            // 如果这哥们没拿到，那就老老实实的走我们的逻辑
            acquire(1);
    }
    
    // 将当前线程赋值于exclusiveOwnerThread
    protected final void setExclusiveOwnerThread(Thread thread) {
        exclusiveOwnerThread = thread;
    }
    复制代码

我们再看下公平锁的代码：

final void lock() {
        // 老老实实的走我们的逻辑
        acquire(1);
    }
    复制代码

会不会突然发现，公平锁和非公平锁就差了一个上来CAS的步骤

我们继续往下看

2.1.1 acquire

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

我们首先看一下 `tryAcquire` 方法，这个方法也分公平锁和非公平锁，但实现差距不大

2.1.2 tryAcquire

**公平锁**

protected final boolean tryAcquire(int acquires) {
        // 拿到当前的线程
        final Thread current = Thread.currentThread();
        // 获取当前AQS的标记位
        int c = getState();
        // 如果发现是0，说明当前没有人拿到锁，我可以直接去拿
        if (c == 0) {
            // hasQueuedPredecessors:当前线程是否符合更改State的资格
            // compareAndSetState:CAS将0修改为1
            if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
                // AQS里面的占领锁的线程给设置成当前请求的线程
                setExclusiveOwnerThread(current);
                return true;
            }
        } else if (current == getExclusiveOwnerThread()) {
            // 如果current == getExclusiveOwnerThread()的话,说明当前线程已经持有锁的，触发可重入功能
            // State加一
            int nextc = c + acquires;
            // 判断一下这哥们是不是越界了(Integer.MAX_VALUE)
            if (nextc < 0){
                throw new Error("Maximum lock count exceeded");
            }
            // 将State设置进去
            setState(nextc);
            return true;
        }
        // 如果即拿不到State=0,又没办法可重入，只能抱歉，再见了您
        return false;
    }
    
    // 若想让我们执行上面的逻辑，这里的返回值必须是false
    // 我们分析一下false的情况：
    // 当前AQS中无数据，h == t，直接返回false
    // 当前AQS中有数据,当前线程等于等待队列中的第一个线程，直接返回false(这里防止你后续排队的线程插队的情况)
    public final boolean hasQueuedPredecessors() {
        Node t = tail; 
        Node h = head;
        Node s = h.next;
        return h != t && (s == null || s.thread != Thread.currentThread());
    }
    复制代码

非公平锁

观察一下这个非公平锁的方法，有莫有什么发现

对了，就和我们的 公平锁 差一个 `hasQueuedPredecessors` 的方法

**人家公平锁来了之后，会先看看自己是不是排队的第一个，如果不是的话，我就不抢占state了**

**但非公平锁不这样想，这哥们来了，我管你是不是排队第一个，我都去CAS一把，看看能不能拿到**

protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
    // 这里的代码合前面一样，我们就不再说了
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (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;
    }
    复制代码

2.1.3 addWaiter

当我们线程能够拿到 `state` 时，直接 `true`，结束即可

但当我们拿不到 `state` 时，就需要继续往下走

`addWaiter` 就是将当前封装成 `Node` 插入到 `AQS` 中，当然插入的时候有两种情况：

*  AQS数据为空：这个时候我们需要初始化完再将当前 `Node` 插入
 *  AQS数据不为空：直接将当前 `Node` 插入即可

// Node EXCLUSIVE：互斥锁
    private Node addWaiter(Node mode) {
        // 哥们你既然拿不到的话，那你就封装成Node排队去吧
        Node node = new Node(Thread.currentThread(), mode);
        // 定义一个指针，指向末尾
        Node pred = tail;
        // 如果指针不为空的话，证明里面是有Node的
        if (pred != null) {
            // 这里主要做的操作是将当前的node插入到队尾
            // 一共分三步
            // 1:将当前node的前继指针指向pred
            // 2:将当前tail的指向node
            // 3:将当前pred的后继指针指向node
            // 最后返回node即可
            // 这里如果搞不明白的，可以自己画图演示一遍
            node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        // 如果当前的pred等于null，也就证明AQS里面没有数据，需要添加
        enq(node);
        return node;
    }
    // 循环插入，一定要将当前node插入到AQS中
    private Node enq(final Node node) {
        for (;;) {
            // 这里再看一下tail的值
            Node t = tail;
            if (t == null) {
                // 如果还是为null的话，说明我们需要初始化了
                // CAS将head指向新创建的【new Node()】
                // tail也指向一下，这样我们初始化就搞好了
                // 记住：这里初始化搞完，还是需要走下面的逻辑，目的是为了将当前的线程封装进去，我们之前封装的是一个虚拟头结点(这也是AQS的灵魂所在)
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                // 到这一步就是这哥们在前面CAS抢占的时候，失败了或者初始化完需要封装当前线程
                // 再这里再重复一遍上面的逻辑
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
    复制代码

2.1.4 acquireQueued

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                // 拿到当前node的前继节点
                final Node p = node.predecessor();
                // 前继节点是head：代表当前节点处于AQS第一个位置
                // tryAcquire返回true:让这哥们再次去尝试一下，能不能拿到锁
                // 因为这里前面没有节点了，自然而然就轮到该节点了
                if (p == head && tryAcquire(arg)) {
                    // 这哥们抢成功之后，AQS中需要把他删除
                    // 还记得我们前面有一个虚拟节点嘛，我们要让node替代我们的虚拟节点
                    // 将head指向node,将node的pre设置为null
                    // 将p的next设置为null,让其没有链接，可达性分析直接GC掉
                    // 这段也需要自己画图演示一下
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                // 如果这哥们前继节点不是head的话
                // 说明我们需要对当前的线程做一个挂起的操作，但是挂起之前我们需要校验一下前继Node的状态
                // 如果当前的前继节点为Node.SIGNAL时，继续执行parkAndCheckInterrupt
                //
                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()){
                    interrupted = true;
                }
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    
    // pred：前继节点
    // node：当前节点
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 拿到前继节点的等待状态
        int ws = pred.waitStatus;
        // 如果前继节点Node.SIGNAL，说明前继Node是好的，直接返回true就可以了
        // 这里观察前继节点的原因是因为：我们当前节点的唤醒操作是需要指望前继节点的
        if (ws == Node.SIGNAL)
            return true;
        // 如果前继节点状态为1的话，证明前继节点获取锁的请求已经取消了
        // 也就是该节点已经无效了
        if (ws > 0) {
            do {
                // 我们当前节点需要自我救赎一下，不能你这个前继节点完蛋了，当前节点也跟着完蛋吧
                // 怎么救赎呢？
                // 我们当前节点一直向前探索，直到找到能用的前继节点为止
                // 然后将pred挂在这个有效的前继节点上，前继节点的next挂到当前线程
                // 这部分操作完成之后，中间无效的节点可直接被GC回收掉
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 如果到这里，说明我们的状态如下：节点线程等待唤醒
            // 那么我们CAS将该节点的状态修改为Node.SIGNAL
            // 这样的话，我们当前节点可以保证前继节点是有效的
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        // 非有效的情况下，均返回false
        return false;
    }
    
    // 直接调用底层park指令将当前线程挂起
    private final boolean parkAndCheckInterrupt() {
        // 这一块会将系统挂起,等到前继节点执行unpark指令将该线程唤醒
        LockSupport.park(this);
        return Thread.interrupted();
    }
    public static void park(Object blocker) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        UNSAFE.park(false, 0L);
        setBlocker(t, null);
    }
    复制代码

2.1.5 selfInterrupt

当我们的 `acquireQueued` 方法执行 `selfInterrupt` 返回 `true` 时，这个时候才会执行我们的 `selfInterrupt` 方法

这个地方特别有意思，博主看了好多文章才参透其中的奥秘

首先，我们看什么时候会返回 `true`

private final boolean parkAndCheckInterrupt() {
        // 这一块会将系统挂起,等到前继节点执行unpark指令将该线程唤醒
        LockSupport.park(this);
        // 这里返回true的时候才会执行selfInterrupt方法
        return Thread.interrupted();
    }
    复制代码

所以，我们先来解释一下 `Thread.interrupted()` 是什么意思以及什么情况下会返回 `true`

对于线程来说，执行中断的标记一共有三个方法：

*  interrupt：是给线程设置中断标志
 *  interrupted：是检测中断并清除中断状态
 *  isInterrupted：只检测中断

所以，我们上面的 `Thread.interrupted()` 返回 `true` 的时机是当前线程处于中断标记，执行之后会清除其中断状态

我们一起来看下 `selfInterrupt` 的执行

static void selfInterrupt() {
        Thread.currentThread().interrupt();
    }
    复制代码

咦，你突然的会发现，我靠，这个地方干嘛又给他设置中断状态了，这么奇怪。

这部分属于Java提供的协作式中断知识内容，这里简单介绍一下：

*  当中断线程被唤醒时，并不知道被唤醒的原因，可能是当前线程在等待中被中断，也可能是释放了锁以后被唤醒。因此我们通过 `Thread.interrupted()` 方法检查中断标记（该方法返回了当前线程的中断状态，并将当前线程的中断标识设置为 `False`），并记录下来，如果发现该线程被中断过，就再中断一次。
 *  线程在等待资源的过程中被唤醒，唤醒后还是会不断地去尝试获取锁，直到抢到锁为止。也就是说，在整个流程中，并不响应中断，只是记录中断记录。最后抢到锁返回了，那么如果被中断过的话，就需要补充一次中断。

关于 `Java协作式中断`，可以参考下：[blog.csdn.net/reliveIT/ar…][blog.csdn.net_reliveIT_ar]

到这里，我们整个的 `lock` 流程就结束了

### 3、unLock 源码 ###

我们释放锁的源码里面没有公平锁和非公平锁区分

// 解锁
    lock.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;
    }
    复制代码

3.1 tryRelease

*  当前该线程是否为占用state的线程，不是则抛出异常
 *  如果是的话则将state减一，看其是否为0（是否仍然有线程持有该锁）

// releases = 1
    protected final boolean tryRelease(int releases) {
        // 获取当前的State减一
        int c = getState() - releases;
        // 比较当前的线程与获取锁的线程是不是同一个
        if (Thread.currentThread() != getExclusiveOwnerThread()){
            // 不是同一个，则抛出异常
            throw new IllegalMonitorStateException();
        }
        
        boolean free = false;
        // 如果当前state为0的话，则代表没有线程持有锁
        if (c == 0) {
            // 将空闲标记置为true
            free = true;
            // 将持有锁的线程设置为null
            setExclusiveOwnerThread(null);
        }
        
        // 设置state
        setState(c);
        // 如果当前没有线程持有state，这里会返回true
        // 如果仍然存在线程持有state，这里会返回false
        return free;
    }
    复制代码

3.2 unparkSuccessor

当我们上面没有线程持有 `state` 时，会走向下面的代码：

public final boolean release(int arg) {
        if (tryRelease(arg)) {
            // 设置h引用指向head
            Node h = head;
            // 当前AQS中有数据并且头指针的节点是有效的
            if (h != null && h.waitStatus != 0)
                // 唤醒节点
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    复制代码

那这个唤醒节点到底是如何操作的呢？我们一起来看一下：

// node = head
    private void unparkSuccessor(Node node) {
        
        // 拿到当前head节点的等待状态
        int ws = node.waitStatus;
        
        // 如果小于0,先CAS将其Status改为0
        // 这里修改的原因：如果我们不修改的话，在并发情况下，这里会频繁的去唤醒后续的节点
        // 有点类似我们之前kafka的mute，唤醒一个我就关闭一下，唤醒一个关闭一下
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
    
       	// 拿到头结点的下一个节点(正式节点)
        Node s = node.next;
        // 如果下一个节点是空的或者节点无效的，需要执行下述操作
        if (s == null || s.waitStatus > 0) {
            // 无效的话，直接将节点置空就可以了
            s = null;
            // 从后往前找到状态小于等于0的节点
            // 找到离head最新的有效节点，并赋值给s
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        // 当前节点不等于空的情况下，直接unpark唤醒即可
        if (s != null)
            LockSupport.unpark(s.thread);
    }
    复制代码

这里简单的来介绍下从后往前找：

上面我们可以知道，当 **`下一个节点是空的或者节点无效的`** 的时候，会触发向前寻找的操作

![format_png 2][]

比如上面这种情况，如果我们继续从头结点开始寻找的话，实际上我们是找不到后面的节点的（在黄色null已经空指针了）

所以，我们需要从 `tail` 指针，从后往前找到符合我们要求（有效）的节点，也就是 `蓝色` 的节点。

然后将其唤醒即可。

## 四、流程图 ##

![format_png 3][]

## 五、写在最后 ##

鲁迅先生曾说：**独行难，众行易，和志同道合的人一起进步。彼此毫无保留的分享经验，才是对抗互联网寒冬的最佳选择。**

[format_png]: https://img-blog.csdnimg.cn/img_convert/ed01c1a2c84e84e87514c854b5fd7f76.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/0796cdf62ac8481863765afb34245c57.webp?x-oss-process=image/format,png
[blog.csdn.net_reliveIT_ar]: https://link.juejin.cn?target=https%3A%2F%2Fblog.csdn.net%2FreliveIT%2Farticle%2Fdetails%2F49920205
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/0cdeed84bd6b962bbac534c6f537f7d8.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/b4fbaab917212d797271c094584ab377.webp?x-oss-process=image/format,png