详解Java ReentrantReadWriteLock读写锁的原理与实现

目录

• 概述
• 原理概述
• 加锁原理
• 图解过程
• 源码解析
• 解锁原理
• 图解过程
• 源码解析

概述

ReentrantReadWriteLock读写锁是使用AQS的集大成者，用了独占模式和共享模式。本文和大家一起理解下ReentrantReadWriteLock读写锁的实现原理。在这之前建议大家阅读下下面3篇关联文章：

深入浅出理解Java并发AQS的独占锁模式

深入浅出理解Java并发AQS的共享锁模式

通俗易懂读写锁ReentrantReadWriteLock的使用

原理概述

上图是ReentrantReadWriteLock读写锁的类结构图：

• 实现了ReadWriteLock接口，该接口提供了获取读锁和写锁的API。
• ReentrantReadWriteLock读写锁内部的成员变量readLock是读锁，指向内部类ReadLock。
• ReentrantReadWriteLock读写锁内部的成员变量writeLock是写锁，指向内部类WriteLock。
• ReentrantReadWriteLock读写锁内部的成员变量sync是继承AQS的同步器，他有两个子类FairSync公平同步器和NoFairSync非公平同步器，读写锁内部也有一个sync，他们使用的是同一个sync。

读写锁用的同一个sync同步器，那么他们共享同一个state, 这样不会混淆吗？

不会，ReentrantReadWriteLock读写锁使用了AQS中state值得低16位表示写锁得计数，用高16位表示读锁得计数，这样就可以使用同一个AQS同时管理读锁和写锁。

1.ReentrantReadWriteLock类重要成员变量

// 读锁
private final ReentrantReadWriteLock.ReadLock readerLock;
// 写锁
private final ReentrantReadWriteLock.WriteLock writerLock;
// 同步器
final Sync sync;

2.ReentrantReadWriteLock构造方法

//默认是非公平锁，可以指定参数创建公平锁
public ReentrantReadWriteLock(boolean fair) {
    // true 为公平锁
    sync = fair ? new FairSync() : new NonfairSync();
    // 这两个 lock 共享同一个 sync 实例，都是由 ReentrantReadWriteLock 的 sync 提供同步实现
    readerLock = new ReadLock(this);
    writerLock = new WriteLock(this);
}

3.Sync类重要成员变量

// 用来移位
static final int SHARED_SHIFT   = 16;
// 高16位的1
static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
// 65535，16个1，代表写锁的最大重入次数
static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
// 低16位掩码：0b 1111 1111 1111 1111，用来获取写锁重入的次数
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

// 获取读写锁的读锁分配的总次数
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
// 写锁（独占）锁的重入次数
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

加锁原理

图解过程

设计一个加锁场景，t1线程加写锁，t2线程加读锁，我们看下它们整个加锁得流程。

1.t1 加写锁w.lock()成功，占了 state 的低 16 位。

• 这里得state分为两部分0_1,0表示高16位的值，1表示低16位的值。
• AQS当前占用线程exclusiveOwnerThread属性指向t1线程。

2.t2线程执行加读锁 r.lock()，尝试获取锁，发现已经被写锁占据了，加锁失败。

3.t2线程被封装成一个共享模式Node.SHARED的节点，加入到AQS的队列中。

4.在阻塞前，t2线程发现自己是队列中的老二，会尝试再次获取读锁，因为t1没有释放，它会失败，然后它会把队列的前驱节点的状态改为-1，然后阻塞自身，也就是t2线程。

• 上面中黄色三角形就是等待状态的值，前驱节点变成-1
• 上面中的灰色表示节点所在的线程阻塞了

5.后面如过有其他线程如t3,t4加读锁或者写锁，由于t1线程没有释放锁，会变成下面的状态。

上面是整个解锁的流程，下面深入源码验证这个流程。

源码解析

1.写锁加锁源码

WriteLock类的lock()方法是加写锁的入口方法。

static final class NonfairSync extends Sync {
    // ... 省略无关代码

    // 外部类 WriteLock 方法, 方便阅读, 放在此处
    public void lock() {
        sync.acquire(1);
    }

    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final void acquire(int arg) {
        if (
            // 尝试获得写锁失败
                !tryAcquire(arg) &&
                        // 将当前线程关联到一个 Node 对象上, 模式为独占模式
                        // 进入 AQS 队列阻塞
                        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
        ) {
            selfInterrupt();
        }
    }

    protected final boolean tryAcquire(int acquires) {
        // 获取当前线程
        Thread current = Thread.currentThread();
        //获得锁的状态
        int c = getState();
        // 获得低 16 位, 代表写锁的 state 计数
        int w = exclusiveCount(c);
         // c不等于0表示加了读锁或者写锁
        if (c != 0) {
            if (
                // c != 0 and w == 0 表示有读锁返回错误，读锁不支持锁升级, 或者
                    w == 0 ||
                        	// w != 0 说明有写锁，写锁的拥有者不是自己，获取失败
                            current != getExclusiveOwnerThread()
            ) {
                // 获得锁失败
                return false;
            }
            // 写锁计数超过低 16 位最大数量, 报异常
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            // 写锁重入, 获得锁成功，没有并发，所以不使用 CAS
            setState(c + acquires);
            return true;
        }
        if (
             // c == 0，说明没有任何锁，判断写锁是否该阻塞，是 false 就尝试获取锁，失败返回 false
                writerShouldBlock() ||
                        // 尝试更改计数失败
                        !compareAndSetState(c, c + acquires)
        ) {
            // 获得锁失败
            return false;
        }
        // 获得锁成功，设置锁的持有线程为当前线程
        setExclusiveOwnerThread(current);
        return true;
    }

    // 非公平锁 writerShouldBlock 总是返回 false, 无需阻塞
    final boolean writerShouldBlock() {
        return false;
    }
    // 公平锁会检查 AQS 队列中是否有前驱节点, 没有(false)才去竞争
    final boolean writerShouldBlock() {
        return hasQueuedPredecessors();
    }
}

• tryAcquire()方法是模板方法，由子类自定义实现获取锁的逻辑。
• 线程如果获取写锁失败的话，通过acquireQueued()方法封装成独占Node加入到AQS队列中。

2.读锁加锁源码

ReadLock类的lock()方法是加读锁的入口方法，调用tryAcquireShared()方法尝试获取读锁，返回负数，失败，加入到队列中。

// 加读锁的方法入口
public void lock() {
    sync.acquireShared(1);
}
public final void acquireShared(int arg) {
    // tryAcquireShared 返回负数, 表示获取读锁失败，加入到队列中
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

tryAcquireShared()方法是一个模板方法，AQS类中定义语义，子类实现，如果返回1，表示获取锁成功，还有剩余资源，返回0表示获取成功，没有剩余资源，返回-1表示失败。

// 尝试以共享模式获取，返回1表示获取锁成功，还有剩余资源,返回0表示获取成功，没有剩余资源，返回-1，表示失败
protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
    int c = getState();
    // exclusiveCount(c) 代表低 16 位, 写锁的 state，成立说明有线程持有写锁
    // 写锁的持有者不是当前线程，则获取读锁失败，【写锁允许降级】
    if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
        return -1;

    // 高 16 位，代表读锁的 state，共享锁分配出去的总次数
    int r = sharedCount(c);
    // 读锁是否应该阻塞
    if (!readerShouldBlock() &&	r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {		// 尝试增加读锁计数
        // 加锁成功
        // 加锁之前读锁为 0，说明当前线程是第一个读锁线程
        if (r == 0) {
            firstReader = current;
            firstReaderHoldCount = 1;
        // 第一个读锁线程是自己就发生了读锁重入
        } else if (firstReader == current) {
            firstReaderHoldCount++;
        } else {
            // cachedHoldCounter 设置为当前线程的 holdCounter 对象，即最后一个获取读锁的线程
            HoldCounter rh = cachedHoldCounter;
            // 说明还没设置 rh
            if (rh == null || rh.tid != getThreadId(current))
                // 获取当前线程的锁重入的对象，赋值给 cachedHoldCounter
                cachedHoldCounter = rh = readHolds.get();
            // 还没重入
            else if (rh.count == 0)
                readHolds.set(rh);
            // 重入 + 1
            rh.count++;
        }
        // 读锁加锁成功
        return 1;
    }
    // 逻辑到这 应该阻塞，或者 cas 加锁失败
    // 会不断尝试 for (;;) 获取读锁, 执行过程中无阻塞
    return fullTryAcquireShared(current);
}

// 非公平锁 readerShouldBlock 偏向写锁一些，看 AQS 阻塞队列中第一个节点是否是写锁，是则阻塞，反之不阻塞
// 防止一直有读锁线程，导致写锁线程饥饿
// true 则该阻塞, false 则不阻塞
final boolean readerShouldBlock() {
    return apparentlyFirstQueuedIsExclusive();
}

// 下面是公平锁的readerShouldBlock
// 公平锁会检查 AQS 队列中是否有前驱节点, 没有(false)才去竞争
final boolean readerShouldBlock() {
    return hasQueuedPredecessors();
}

fullTryAcquireShared()方法是通过自旋的方式不断获取读锁，因为由于前面的readerShouldBlock返回false或者cas失败，导致没有获取到锁，需要不断重试。

final int fullTryAcquireShared(Thread current) {
    // 当前读锁线程持有的读锁次数对象
    HoldCounter rh = null;
    for (;;) {
        int c = getState();
        // 说明有线程持有写锁
        if (exclusiveCount(c) != 0) {
            // 写锁不是自己则获取锁失败
            if (getExclusiveOwnerThread() != current)
                return -1;
        } else if (readerShouldBlock()) {
            // 条件成立说明当前线程是 firstReader，当前锁是读忙碌状态，而且当前线程也是读锁重入
            if (firstReader == current) {
                // assert firstReaderHoldCount > 0;
            } else {
                if (rh == null) {
                    // 最后一个读锁的 HoldCounter
                    rh = cachedHoldCounter;
                    // 说明当前线程也不是最后一个读锁
                    if (rh == null || rh.tid != getThreadId(current)) {
                        // 获取当前线程的 HoldCounter
                        rh = readHolds.get();
                        // 条件成立说明 HoldCounter 对象是上一步代码新建的
                        // 当前线程不是锁重入，在 readerShouldBlock() 返回 true 时需要去排队
                        if (rh.count == 0)
                            // 防止内存泄漏
                            readHolds.remove();
                    }
                }
                if (rh.count == 0)
                    return -1;
            }
        }
        // 越界判断
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 读锁加锁，条件内的逻辑与 tryAcquireShared 相同
        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;
        }
    }
}

doAcquireShared()是在获取读锁失败的时候加入AQS队列的逻辑。

private void doAcquireShared(int arg) {
    // 将当前线程关联到一个 Node 对象上, 模式为共享模式
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            // 获取前驱节点
            final Node p = node.predecessor();
            // 如果前驱节点就头节点就去尝试获取锁
            if (p == head) {
                // 再一次尝试获取读锁
                int r = tryAcquireShared(arg);
                // r >= 0 表示获取成功
                if (r >= 0) {
                    //【这里会设置自己为头节点，唤醒相连的后序的共享节点】
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            // 是否在获取读锁失败时阻塞      					 park 当前线程
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

setHeadAndPropagate()方法是在后续读锁被唤醒后，抢到锁要处理的逻辑，包括修改队列的头结点，以及唤醒队列中的下一个共享节点。

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head;
    // 设置自己为 head 节点
    setHead(node);
    // propagate 表示有共享资源（例如共享读锁或信号量），为 0 就没有资源
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        // 获取下一个节点
        Node s = node.next;
        // 如果当前是最后一个节点，或者下一个节点是【等待共享读锁的节点】
        if (s == null || s.isShared())
            // 唤醒后继节点
            doReleaseShared();
    }
}

解锁原理

图解过程

由于上面t1线程加的写锁，所有其他的线程都被阻塞了，只有在t1线程解锁以后，其他线程才能被唤醒，我们现在看下t1线程被唤醒了，会发生什么？

1.t1线程执行解锁w.unlock()成功，修改AQS中的state。

• 这里的state变为了0_0。
• AQS当前占用线程exclusiveOwnerThread属性变为null。

2.t1线程唤醒队列中等待的老二， 为什么不是老大，因为老大是一个空节点，不会设置任何的线程。t2线程被唤醒后，抢锁成功，修改state中高16位为1。

• 老二的线程节点变为蓝色节点
• AQS中的state变为1_0。

3.t2线程恢复运行，设置原来的老二节点为头节点

4.t2线程要做的事情还没结束呢，因为是共享模式，它现在释放了，就此时也唤醒队列中的下一个共享节点。

5.t3线程恢复去竞争读锁成功，这时state的高位+1，变成2。

6.这时候t3线程所在的Node设置为头节点，同时发现对列的下一个节点不是共享节点，而是独占节点，就不会唤醒后面的节点了。

7.之后t2线程和t3线程进入尾声，执行r.unlock操作，state的计数减一，直到变为0。

8.最后写锁线程t4被唤醒，去抢占锁成功，整个流程结束。

上面是整个解锁的流程，下面深入源码验证这个流程。

源码解析

1.写锁释放流程

WriteLock类的unlock()方法是入口方法，调用tryRelease()方法释放锁，如果成功，调用unparkSuccessor()方法唤醒线程。

public void unlock() {
    // 释放锁
    sync.release(1);
}
public final boolean release(int arg) {
    // 尝试释放锁
    if (tryRelease(arg)) {
        Node h = head;
        // 头节点不为空并且不是等待状态不是 0，唤醒后继的非取消节点
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

tryRelease()方法是AQS提供的模板方法，返回true表示成功，false失败，由自定义同步器实现。

protected final boolean tryRelease(int releases) {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    int nextc = getState() - releases;
    // 因为可重入的原因, 写锁计数为 0, 才算释放成功
    boolean free = exclusiveCount(nextc) == 0;
    if (free)
        // 设置占用线程为null
        setExclusiveOwnerThread(null);
    setState(nextc);
    return free;
}

2.读锁释放流程

ReadLock类的unlock()方法是释放共享锁的入口方法。

public void unlock() {
    sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

tryReleaseShared()方法是由AQS提供的模板方法，由自定义同步器实现。

protected final boolean tryReleaseShared(int unused) {
    //自选
    for (;;) {
        int c = getState();
        int nextc = c - SHARED_UNIT;
        // 读锁的计数不会影响其它获取读锁线程, 但会影响其它获取写锁线程，计数为 0 才是真正释放
        if (compareAndSetState(c, nextc))
            // 返回是否已经完全释放了
            return nextc == 0;
    }
}

调用doReleaseShared()方法唤醒等待的线程，这个方法调用的地方有两处，还记得吗，一个这是里的解锁，还有一个是前面加共享锁阻塞的地方，唤醒后获取锁成功，也会调用doReleaseShared()方法。

private void doReleaseShared() {
    // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
	// 如果 head.waitStatus == 0 ==> Node.PROPAGATE
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            // SIGNAL 唤醒后继
            if (ws == Node.SIGNAL) {
                // 因为读锁共享，如果其它线程也在释放读锁，那么需要将 waitStatus 先改为 0
            	// 防止 unparkSuccessor 被多次执行
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                // 唤醒后继节点
                unparkSuccessor(h);
            }
            // 如果已经是 0 了，改为 -3，用来解决传播性
            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        // 条件不成立说明被唤醒的节点非常积极，直接将自己设置为了新的 head，
        // 此时唤醒它的节点（前驱）执行 h == head 不成立，所以不会跳出循环，会继续唤醒新的 head 节点的后继节点
        if (h == head)
            break;
    }
}

以上就是详解Java ReentrantReadWriteLock读写锁的原理与实现的详细内容，更多关于Java ReentrantReadWriteLock读写锁的资料请关注我们其它相关文章！