Java 高并发编程：深入解析 ReentrantReadWriteLock 锁实现

写锁重入：读写锁中的写锁的重入方式基本和 ReentrantLock 一致，没有什么区别，依然是对 state 进行 +1 操作即可，只要确认持有锁资源的线程是当前写锁线程即可。只不过之前 ReentrantLock 的重入次数是 state 的正数取值范围，但是读写锁中写锁范围就变小了。

读锁重入：因为读锁是共享锁。读锁在获取锁资源操作时，是要对 state 的高 16 位进行 +1 操作。因为读锁是共享锁，所以同一时间会有多个读线程持有读锁资源。这样一来，多个读操作在持有读锁时，无法确认每个线程读锁重入的次数。为了去记录读锁重入的次数，每个读操作的线程，都会有一个 ThreadLocal 记录锁重入的次数。

写锁的饥饿问题：读锁是共享锁，当有线程持有读锁资源时，再来一个线程想要获取读锁，直接对 state 修改即可。在读锁资源先被占用后，来了一个写锁资源，此时，大量的需要获取读锁的线程来请求锁资源，如果可以绕过写锁，直接拿资源，会造成写锁长时间无法获取到写锁资源。

读锁在拿到锁资源后，如果再有读线程需要获取读锁资源，需要去 AQS 队列排队。如果队列的前面需要写锁资源的线程，那么后续读线程是无法拿到锁资源的。持有读锁的线程，只会让写锁线程之前的读线程拿到锁资源。

4.3 写锁分析

4.3.1 写锁加锁流程概述

（此处省略流程图）

4.3.2 写锁加锁源码分析

写锁加锁流程

// 写锁加锁的入口
public void lock() {
    sync.acquire(1);
}

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

// 读写锁的写锁实现
protected final boolean tryAcquire(int acquires) {
    // 拿到当前线程
    Thread current = Thread.currentThread();
    // 拿到 state 的值
    int c = getState();
    // 得到 state 低 16 位的值
    int w = exclusiveCount(c);
    // 判断是否有线程持有着锁资源
    if (c != 0) {
        // 当前没有线程持有写锁，读写互斥，告辞。
        // 有线程持有写锁，持有写锁的线程不是当前线程，不是锁重入，告辞。
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;
        // 当前线程持有写锁。锁重入。
        if (w + exclusiveCount(acquires) > MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 没有超过锁重入的次数，正常 +1
        setState(c + acquires);
        return true;
    }
    // 尝试获取锁资源
    if (writerShouldBlock() || !compareAndSetState(c, c + acquires))
        return false;
    // 拿锁成功，设置占有互斥锁的线程
    setExclusiveOwnerThread(current);
    // 返回 true
    return true;
}

4.3.3 写锁释放锁流程概述 & 释放锁源码

释放的流程和 ReentrantLock 一致，只是在判断释放是否干净时，判断低 16 位的值。

// 写锁释放锁的 tryRelease 方法
protected final boolean tryRelease(int releases) {
    // 判断当前持有写锁的线程是否是当前线程
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 获取 state - 1
    int nextc = getState() - releases;
    // 判断低 16 位结果是否为 0，如果为 0，free 设置为 true
    boolean free = exclusiveCount(nextc) == 0;
    if (free)
        // 将持有锁的线程设置为 null
        setExclusiveOwnerThread(null);
    // 设置给 state
    setState(nextc);
    // 释放干净，返回 true。写锁有冲入，这里需要返回 false，不去释放排队的 Node
    return free;
}

4.4 读锁分析

4.4.1 读锁加锁流程概述

1. 分析读锁加速的基本流程
2. 分析读锁的可重入锁实现以及优化
3. 解决 ThreadLocal 内存泄漏问题
4. 读锁获取锁自后，如果唤醒 AQS 中排队的读线程

4.4.1.1 基础读锁流程

针对上述简单逻辑的源码分析

// 读锁加锁的方法入口
public final void acquireShared(int arg) {
    // 竞争锁资源滴干活
    if (tryAcquireShared(arg) < 0)
        // 没拿到锁资源，去排队
        doAcquireShared(arg);
}

// 读锁竞争锁资源的操作
protected final int tryAcquireShared(int unused) {
    // 拿到当前线程
    Thread current = Thread.currentThread();
    // 拿到 state
    int c = getState();
    // 拿到 state 的低 16 位，判断 != 0，有写锁占用着锁资源
    // 并且，当前占用锁资源的线程不是当前线程
    if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
        // 写锁被其他线程占用，无法获取读锁，直接返回 -1，去排队
        return -1;
    // 没有线程持有写锁、当前线程持有写锁
    // 获取读锁的信息，state 的高 16 位。
    int r = sharedCount(c);
    // 公平锁：就查看队列是由有排队的，有排队的，直接告辞，进不去 if，后面也不用判断（没人排队继续走）
    // 非公平锁：没有排队的，直接抢。有排队的，但是读锁其实不需要排队，如果出现这个情况，大部分是写锁资源刚刚释放，
    // 后续 Node 还没有来记得拿到读锁资源，当前竞争的读线程，可以直接获取
    if (!readerShouldBlock() &&
        // 判断持有读锁的临界值是否达到 r < MAX_COUNT
        r < MAX_COUNT &&
        // CAS 修改 state，对高 16 位进行 +1
        compareAndSetState(c, c + SHARED_UNIT)) {
        // 省略部分代码！！！！
        return 1;
    }
    return fullTryAcquireShared(current);
}

// 非公平锁的判断
final boolean apparentlyFirstQueuedIsExclusive() {
    Node h, s;
    return (h = head) != null &&
           // head 为 null，可以直接抢占锁资源
           (s = h.next) != null &&
           // head 的 next 为 null，可以直接抢占锁资源
           !s.isShared() &&
           // 如果排在 head 后面的 Node，是共享锁，可以直接抢占锁资源。s.thread !=null；
           // 后面排队的 thread 为 null，可以直接抢占锁资源
           s.thread != null;
}

4.4.1.2 读锁重入流程

\=============重入操作

前面阐述过，读锁为了记录锁重入的次数，需要让每个读线程用 ThreadLocal 存储重入次数

ReentrantReadWriteLock 对读锁重入做了一些优化操作

\============记录重入次数的核心

ReentrantReadWriteLock 在内部对 ThreadLocal 做了封装，基于 HoldCount 的对象存储重入次数，在内部有个 count 属性记录，而且每个线程都是自己的 ThreadLocalHoldCounter，所以可以直接对内部的 count 进行++操作。

\=============第一个获取读锁资源的重入次数记录方式

第一个拿到读锁资源的线程，不需要通过 ThreadLocal 存储，内部提供了两个属性来记录第一个拿到读锁资源线程的信息

内部提供了 firstReader 记录第一个拿到读锁资源的线程，firstReaderHoldCount 记录 firstReader 的锁重入次数

\==============最后一个获取读锁资源的重入次数记录方式

最后一个拿到读锁资源的线程，也会缓存他的重入次数，这样++起来更方便

基于 cachedHoldCounter 缓存最后一个拿到锁资源现行的重入次数

\==============最后一个获取读锁资源的重入次数记录方式

重入次数的流程执行方式：

1、判断当前线程是否是第一个拿到读锁资源的：如果是，直接将 firstReader 以及 firstReaderHoldCount 设置为当前线程的信息

2、判断当前线程是否是 firstReader：如果是，直接对 firstReaderHoldCount++ 即可。

3、跟 firstReader 没关系了，先获取 cachedHoldCounter，判断是否是当前线程。

3.1、如果不是，获取当前线程的重入次数，将 cachedHoldCounter 设置为当前线程。

3.2、如果是，判断当前重入次数是否为 0，重新设置当前线程的锁从入信息到 readHolds（ThreadLocal）中，算是初始化操作，重入次数是 0

3.3、前面两者最后都做 count++

上述逻辑源码分析

protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
    int c = getState();
    if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
        return -1;
    int r = sharedCount(c);
    if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) {
        // ===============================================================
        // 判断 r == 0，当前是第一个拿到读锁资源的线程
        if (r == 0) {
            // 将 firstReader 设置为当前线程
            firstReader = current;
            // 将 count 设置为 1
            firstReaderHoldCount = 1;
        }
        // 判断当前线程是否是第一个获取读锁资源的线程
        else if (firstReader == current) {
            // 直接++。
            firstReaderHoldCount++;
        }
        // 到这，就说明不是第一个获取读锁资源的线程
        else {
            // 那获取最后一个拿到读锁资源的线程
            HoldCounter rh = cachedHoldCounter;
            // 判断当前线程是否是最后一个拿到读锁资源的线程
            if (rh == null || rh.tid != getThreadId(current))
                // 如果不是，设置当前线程为 cachedHoldCounter
                cachedHoldCounter = rh = readHolds.get();
            // 当前线程是之前的 cacheHoldCounter
            else if (rh.count == 0)
                // 将当前的重入信息设置到 ThreadLocal 中
                readHolds.set(rh);
            // 重入的++
            rh.count++;
        }
        // ===============================================================
        return 1;
    }
    return fullTryAcquireShared(current);
}

4.4.1.3 读锁加锁的后续逻辑 fullTryAcquireShared

// tryAcquireShard 方法中，如果没有拿到锁资源，走这个方法，尝试再次获取，逻辑跟上面基本一致。
final int fullTryAcquireShared(Thread current) {
    // 声明当前线程的锁重入次数
    HoldCounter rh = null;
    // 死循环
    for (;;) {
        // 再次拿到 state
        int c = getState();
        // 当前如果有写锁在占用锁资源，并且不是当前线程，返回 -1，走排队策略
        if (exclusiveCount(c) != 0) {
            if (getExclusiveOwnerThread() != current)
                return -1;
        }
        // 查看当前是否可以尝试竞争锁资源（公平锁和非公平锁的逻辑）
        else if (readerShouldBlock()) {
            // 无论公平还是非公平，只要进来，就代表要放到 AQS 队列中了，先做一波准备
            // 在处理 ThreadLocal 的内存泄漏问题
            if (firstReader == current) {
                // 如果当前当前线程是之前的 firstReader，什么都不用做
            } else {
                // 第一次进来是 null。
                if (rh == null) {
                    // 拿到最后一个获取读锁的线程
                    rh = cachedHoldCounter;
                    // 当前线程并不是 cachedHoldCounter，没到拿到
                    if (rh == null || rh.tid != getThreadId(current)) {
                        // 从自己的 ThreadLocal 中拿到重入计数器
                        rh = readHolds.get();
                        // 如果计数器为 0，说明之前没拿到过读锁资源
                        if (rh.count == 0)
                            // remove，避免内存泄漏
                            readHolds.remove();
                    }
                }
                // 前面处理完之后，直接返回 -1
                if (rh.count == 0)
                    return -1;
            }
        }
        // 判断重入次数，是否超出阈值
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // CAS 尝试获取锁资源
        if (compareAndSetState(c, c + SHARED_UNIT)) {
            if (sharedCount(c) == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                if (rh == null)
                    rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
                cachedHoldCounter = rh; // cache for release
            }
            return 1;
        }
    }
}

4.4.1.4 读线程在 AQS 队列获取锁资源的后续操作

1、正常如果都是读线程来获取读锁资源，不需要使用到 AQS 队列的，直接 CAS 操作即可

2、如果写线程持有着写锁，这是读线程就需要进入到 AQS 队列排队，可能会有多个读线程在 AQS 中。

当写锁释放资源后，会唤醒 head 后面的读线程，当 head 后面的读线程拿到锁资源后，还需要查看 next 节点是否也是读线程在阻塞，如果是，直接唤醒

源码分析

// 读锁需要排队的操作
private void doAcquireShared(int arg) {
    // 声明 Node，类型是共享锁，并且扔到 AQS 中排队
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            // 拿到上一个节点
            final Node p = node.predecessor();
            // 如果 prev 节点是 head，直接可以执行 tryAcquireShared
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 拿到读锁资源后，需要做的后续处理
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            // 找到 prev 有效节点，将状态设置为 -1，挂起当前线程
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

private void setHeadAndPropagate(Node node, int propagate) {
    // 拿到 head 节点
    Node h = head;
    // 将当前节点设置为 head 节点
    setHead(node);
    // 第一个判断更多的是在信号量有处理 JDK1.5 BUG 的操作。
    if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
        // 拿到当前 Node 的 next 节点
        Node s = node.next;
        // 如果 next 节点是共享锁，直接唤醒 next 节点
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

4.4.2 读锁的释放锁流程

1、处理重入以及 state 的值 2、唤醒后续排队的 Node

源码分析

// 读锁释放锁流程
public final boolean releaseShared(int arg) {
    // tryReleaseShared：处理 state 的值，以及可重入的内容
    if (tryReleaseShared(arg)) {
        // AQS 队列的事！
        doReleaseShared();
        return true;
    }
    return false;
}

// 1、处理重入问题 2、处理 state
protected final boolean tryReleaseShared(int unused) {
    // 拿到当前线程
    Thread current = Thread.currentThread();
    // 如果是 firstReader，直接干活，不需要 ThreadLocal
    if (firstReader == current) {
        assert firstReaderHoldCount > 0;
        if (firstReaderHoldCount == 1)
            firstReader = null;
        else
            firstReaderHoldCount--;
    }
    // 不是 firstReader，从 cachedHoldCounter 以及 ThreadLocal 处理
    else {
        // 如果是 cachedHoldCounter，正常--
        HoldCounter rh = cachedHoldCounter;
        // 如果不是 cachedHoldCounter，从自己的 ThreadLocal 中拿
        if (rh == null || rh.tid != getThreadId(current))
            rh = readHolds.get();
        int count = rh.count;
        // 如果为 1 或者更小，当前线程就释放干净了，直接 remove，避免 value 内存泄漏
        if (count <= 1) {
            readHolds.remove();
            // 如果已经是 0，没必要再 unlock，扔个异常
            if (count <= 0)
                throw unmatchedUnlockException();
        }
        // -- 走你。
        rh.count--;
    }
    for (;;) {
        // 拿到 state，高 16 位，-1，成功后，返回 state 是否为 0
        int c = getState();
        int nextc = c - SHARED_UNIT;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

// 唤醒 AQS 中排队的线程
private void doReleaseShared() {
    // 死循环
    for (;;) {
        // 拿到头
        Node h = head;
        // 说明有排队的
        if (h != null && h != tail) {
            // 拿到 head 的状态
            int ws = h.waitStatus;
            // 判断是否为 -1
            if (ws == Node.SIGNAL) {
                // 到这，说明后面有挂起的线程，先基于 CAS 将 head 的状态从 -1，改为 0
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                // 唤醒后续节点
                unparkSuccessor(h);
            }
            // 这里不是给读写锁准备的，在信号量里说。。。
            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        // 这里是出口
        if (h == head)
            break;
    }
}