ReentrantReadWriteLock 底层原理

    技术2022-07-12  77

    文章目录

    1.读写锁自定义2. ReentrantReadWriteLock底层原理2.1. t1 w.lock,t2 r.lock2.1.1. 写锁上锁流程(跟ReentrantLock一样,t1 w.lock是独占锁/排它锁)2.1.2. 读锁上锁流程(t2 r.lock共享锁) 2.2 t3 r.lock,t4 w.lock2.3 写锁释放(t1 w.unlock)2.3.1 写锁释放流程及读锁加锁流程 2.4 t2 r.unlock,t3 r.unlock2.4.1 读锁释放流程与写锁加锁流程 3. ReentrantReadWriteLock锁降级与升级

    1.读写锁自定义

    当读操作远远高于写操作时,这时候使用 读写锁 让 读-读 可以并发,提高性能。 类似于数据库中的 select ... from ... lock in share mode提供一个 数据容器类 内部分别使用读锁保护数据的 read() 方法,写锁保护数据的 write()方法

    自定义数据容器类

    class DataContainer { private Object data; private ReentrantReadWriteLock rw = new ReentrantReadWriteLock(); private ReentrantReadWriteLock.ReadLock r = rw.readLock(); private ReentrantReadWriteLock.WriteLock w = rw.writeLock(); public Object read() { log.debug("获取读锁..."); r.lock(); try { log.debug("读取"); sleep(1); return data; } finally { log.debug("释放读锁..."); r.unlock(); } } public void write() { log.debug("获取写锁..."); w.lock(); try { log.debug("写入"); sleep(1); } finally { log.debug("释放写锁..."); w.unlock(); } } }

    测试 读锁-读锁 可以并发,读锁没有释放时,其他线程就可以获取读锁

    DataContainer dataContainer = new DataContainer(); new Thread(() -> { dataContainer.read(); }, "t1").start(); new Thread(() -> { dataContainer.read(); }, "t2").start() 结果:输出结果,从这里可以看到 Thread-0 锁定期间, Thread-1 的读操作不受影响 14:05:14.341 c.DataContainer [t2] - 获取读锁... 14:05:14.341 c.DataContainer [t1] - 获取读锁... 14:05:14.345 c.DataContainer [t1] - 读取 14:05:14.345 c.DataContainer [t2] - 读取 14:05:15.365 c.DataContainer [t2] - 释放读锁... 14:05:15.386 c.DataContainer [t1] - 释放读锁...

    测试 读锁-写锁 相互阻塞,一方线程要等另一方线程锁释放

    DataContainer dataContainer = new DataContainer(); new Thread(() -> { dataContainer.read(); }, "t1").start(); Thread.sleep(100); new Thread(() -> { dataContainer.write(); }, "t2").start(); 结果: 14:04:21.838 c.DataContainer [t1] - 获取读锁... 14:04:21.838 c.DataContainer [t2] - 获取写锁... 14:04:21.841 c.DataContainer [t2] - 写入 14:04:22.843 c.DataContainer [t2] - 释放写锁... 14:04:22.843 c.DataContainer [t1] - 读取 14:04:23.843 c.DataContainer [t1] - 释放读锁...

    写锁-写锁 也是相互阻塞的,必须要等一方线程释放锁,下一个线程才能操作,这里不作测试了


    注意事项

    读锁不支持条件变量重入时升级不支持:即持有读锁的情况下去获取写锁,会导致获取写锁永久等待 r.lock();//读锁 try { // ... w.lock();//写锁永久等待 try { // ... } finally{ w.unlock(); } } finally{ r.unlock(); } 重入时降级支持:即持有写锁的情况下去获取读锁(这是同一个线程重入锁可以) class CachedData { Object data; // 是否有效,如果失效,需要重新计算 data volatile boolean cacheValid; final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(); void processCachedData() { rwl.readLock().lock(); if (!cacheValid) { // 获取写锁前必须释放读锁 rwl.readLock().unlock(); rwl.writeLock().lock(); try { // 判断是否有其它线程已经获取了写锁、更新了缓存, 避免重复更新 if (!cacheValid) { data = ... cacheValid = true; } // 降级为读锁, 释放写锁, 这样能够让其它线程读取缓存 rwl.readLock().lock(); } finally { rwl.writeLock().unlock(); } } // 自己用完数据, 释放读锁 try { use(data); } finally { rwl.readLock().unlock(); } } }

    2. ReentrantReadWriteLock底层原理

    读写锁用的是同一个 Sycn 同步器,因此等待队列、state 等也是同一个

    2.1. t1 w.lock,t2 r.lock

    ReentrantReadWriteLock类结构:

    2.1.1. 写锁上锁流程(跟ReentrantLock一样,t1 w.lock是独占锁/排它锁)

    1). acquire(int arg)方法

    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(); } } }

    2). tryAcquire(arg)方法(子类的实现方法)

    t1 成功上锁,流程与 ReentrantLock 加锁相比没有特殊之处, 不同是写锁状态占了 state 的低 16 位,而读锁使用的 是 state 的高 16 位

    // Sync 继承过来的方法, 方便阅读, 放在此处 protected final boolean tryAcquire(int acquires) { // 获得低 16 位, 代表写锁的 state 计数 Thread current = Thread.currentThread(); int c = getState();//获取写锁当前的同步状态 //exclusiveCount(c)获取写锁获取的次数 int w = exclusiveCount(c); if (c != 0) { if ( // c != 0 and w == 0 表示有读锁, 或者 w == 0 || // 如果 exclusiveOwnerThread 不是自己 current != getExclusiveOwnerThread() ) { // 获得锁失败 return false; } // 写锁计数超过低 16 位, 报异常 if (w + exclusiveCount(acquires) > MAX_COUNT) throw new Error("Maximum lock count exceeded"); // 写锁重入, 获得锁成功 setState(c + acquires); return true; } if ( // 判断写锁是否该阻塞, 或者 writerShouldBlock() || // 尝试更改计数失败 !compareAndSetState(c, c + acquires) ) { // 获得锁失败 return false; } // 获得锁成功 setExclusiveOwnerThread(current); return true; }

    3)exclusiveCount( c)方法:写锁被获取的次数

    其中EXCLUSIVE_MASK为: static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; EXCLUSIVE _MASK为1左移16位然后减1,即为0x0000FFFF。而exclusiveCount方法是将同步状态(state为int类型)与0x0000FFFF相与,即取同步状态的低16位。那么低16位代表什么呢?根据exclusiveCount方法的注释为独占式获取的次数即写锁被获取的次数,现在就可以得出来一个结论同步状态的低16位用来表示写锁的获取次数

    static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

    4)writerShouldBlock()方法:判断写锁是否该阻塞 该方法是Sync类中的抽象方法,有公平锁和非公平锁两种实现方式: 对于非公平锁:

    static final class NonfairSync extends Sync { //对于非公平锁总是返回false,不需要阻塞 final boolean writerShouldBlock() { return false; } }

    对于公平锁:

    static final class FairSync extends Sync { //对于公平锁,需要判断 final boolean writerShouldBlock() { return hasQueuedPredecessors(); } }

    5)如果尝试获取锁tryAcquire失败,就进入 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法,与ReentrantLock一样,这里不作细述


    2.1.2. 读锁上锁流程(t2 r.lock共享锁)

    1) acquireShared(arg)方法

    static final class NonfairSync extends Sync { // ReadLock 方法, 方便阅读, 放在此处 public void lock() { sync.acquireShared(1); } // AQS 继承过来的方法, 方便阅读, 放在此处 public final void acquireShared(int arg) { // tryAcquireShared 返回负数, 表示获取读锁失败 if (tryAcquireShared(arg) < 0) { doAcquireShared(arg); } } }

    2) tryAcquireShared(arg)方法

    t2 执行 r.lock,这时进入读锁的 sync.acquireShared(1) 流程,首先会进入 tryAcquireShared 流程。如果有写锁占据 并且获取写锁的线程不是当前线程,那么 tryAcquireShared 返回 -1 表示失败。 tryAcquireShared 返回值表示 * -1 表示失败 * 0 表示成功,但后继节点不会继续唤醒 * 正数表示成功,而且数值是还有几个后继节点需要唤醒,读写锁返回 1

    // Sync 继承过来的方法, 方便阅读, 放在此处 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) ) { // ... 省略不重要的代码 return 1; } return fullTryAcquireShared(current); }

    3) sharedCount( c) 读锁被获取的次数 该方法是获取读锁被获取的次数,是将同步状态(int c)右移16次,即取同步状态的高16位,现在我们可以得出另外一个结论同步状态的高16位用来表示读锁被获取的次数。读写锁是怎样实现分别记录读锁和写锁的状态的,就是通过获取读写锁的次数,和exclusiveCount(int c)获取写锁次数一样

    static int sharedCount(int c) { return c >>> SHARED_SHIFT; }

    4)readerShouldBlock()方法 判断读锁是否该阻塞 这个方法对于公平锁和非公平锁的实现是不同的,也就导致了ReentrantReadWriteLock()对于公平和非公平的两种不同实现:

    对于非公平锁:

    static final class NonfairSync extends Sync { //... final boolean readerShouldBlock() { /** 看 AQS 队列中第一个节点是否是写锁,true 则该阻塞, false 则不阻塞: 由于非公平的竞争,并且读锁可以共享,所以可能会出现源源不断的读,使得写锁永远竞争不到,然后出现饿死的现象(读-读可以共享,读-写阻塞需要等待释放) 通过这个策略,当一个写锁出现在头结点后面的时候,会立刻阻塞所有还未获取读锁的其他线程,让步给写线程先执行(写-读阻塞) */ return apparentlyFirstQueuedIsExclusive(); } }

    公平锁:

    static final class FairSync extends Sync { //... final boolean readerShouldBlock() { //对于公平锁来说,如果有前驱(也就是非头结点),都会进行等待,不允许竞争锁 return hasQueuedPredecessors(); } }

    5) fullTryAcquireShared(current)

    // AQS 继承过来的方法, 方便阅读, 放在此处 // 与 tryAcquireShared 功能类似, 但会不断尝试 for (;;) 获取读锁, 执行过程中无阻塞 final int fullTryAcquireShared(Thread current) { HoldCounter rh = null; for (;;) { int c = getState(); if (exclusiveCount(c) != 0) { if (getExclusiveOwnerThread() != current) return -1; } else if (readerShouldBlock()) { // ... 省略不重要的代码 } if (sharedCount(c) == MAX_COUNT) throw new Error("Maximum lock count exceeded"); if (compareAndSetState(c, c + SHARED_UNIT)) { // ... 省略不重要的代码 return 1; } } }

    如果获取读锁获取失败,就会继续执行下面的doAcquireShared(arg)方法:想象成acquireQueued()方法

    6) doAcquireShared(arg)方法

    1)如果t2线程获取锁失败,这时会进入doAcquireShared(1) 流程, 首先也是调用 addWaiter 添加节点,不同之处在于节点被设置为 Node.SHARED 模式而非 Node.EXCLUSIVE 模式, 注意此时 t2 仍处于活跃状态 。

    2)t2 会看看自己的节点是不是老二,如果是,还会再次调用 tryAcquireShared(1) 来尝试获取锁 3)如果没有成功,在 doAcquireShared 内 for (;;) 循环一次, 把前驱节点的 waitStatus 改为 -1,再 for (;;) 循环一 次尝试 tryAcquireShared(1) 如果还不成功, 那么在 parkAndCheckInterrupt() 处 park

    // 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); // 成功 if (r >= 0) { // ㈠ // r 表示可用资源数, 在这里总是 1 允许传播 //(唤醒 AQS 中下一个 Share 节点) setHeadAndPropagate(node, r); p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; } } if ( // 是否在获取读锁失败时阻塞(前一个阶段 waitStatus == Node.SIGNAL) shouldParkAfterFailedAcquire(p, node) && // park 当前线程 parkAndCheckInterrupt() ) { interrupted = true; } } } finally { if (failed) cancelAcquire(node); } }

    2.2 t3 r.lock,t4 w.lock

    这种状态下,假设又有 t3 加读锁和 t4 加写锁,这 期间 t1 仍然持有锁,就变成了下面的样子

    2.3 写锁释放(t1 w.unlock)

    2.3.1 写锁释放流程及读锁加锁流程

    1、release()方法

    static final class NonfairSync extends Sync { // ... 省略无关代码 // WriteLock 方法, 方便阅读, 放在此处 public void unlock() { sync.release(1); } // AQS 继承过来的方法, 方便阅读, 放在此处 public final boolean release(int arg) { // 尝试释放写锁成功 if (tryRelease(arg)) { // unpark AQS 中等待的线程 Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }

    2、tryRelease()方法

    // Sync 继承过来的方法, 方便阅读, 放在此处 protected final boolean tryRelease(int releases) { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); int nextc = getState() - releases; // 因为可重入的原因, 写锁计数为 0, 才算释放成功 boolean free = exclusiveCount(nextc) == 0; if (free) { setExclusiveOwnerThread(null); } setState(nextc); return free; } } 这时会走到写锁的 sync.release(1) 流程,调用 sync.tryRelease(1) 成功,变成下面的样子 :

    3、unparkSuccessor ()方法

    private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) //将当前线程的节点状态置0 compareAndSetWaitStatus(node, ws, 0); //找到下一个需要唤醒的结点s Node s = node.next; if (s == null || s.waitStatus > 0) { s = null; //如果该节点已经取消获取锁,那就从队尾开始向前找,找到第一个ws<=0的节点,并赋值给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); } 接下来执行唤醒流程sync.unparkSuccessor,即让老二恢复运行:

    4、doAcquireShared()方法 这里开始唤醒读锁,加锁了

    这时 t2 在doAcquireShared 内parkAndCheckInterrupt()处恢复运行, 这回再来一次 for (;;) 执行 tryAcquireShared 成功则让读锁计数加一

    private void doAcquireShared(int arg) { final Node node = addWaiter(Node.SHARED); boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); if (p == head) { //2、继续尝试获取锁资源,让读锁计数加1 int r = tryAcquireShared(arg); if (r >= 0) { //3、唤醒下一个线程 setHeadAndPropagate(node, r); p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; } } //1、t2线程在这儿被唤醒,就会继续指向一次for循环 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }

    5、setHeadAndPropagate (node, 1)方法

    这时 t2 已经恢复运行,接下来 t2 调用 setHeadAndPropagate(node, 1), 它原本所在节点被置为头节点

    private void setHeadAndPropagate(Node node, int propagate) { Node h = head; setHead(node);//head指向自己 //如果锁计数>0,就继续唤醒下面的线程 if (propagate > 0 || h == null || h.waitStatus < 0) { Node s = node.next; //检查下一个节点是否是 shared,如果是将 head 的状态从 -1 改为 0 并唤醒老二 if (s == null || s.isShared()) doReleaseShared(); } } 事情还没完,在setHeadAndPropagate方法内还会检查下一个节点是否是 shared,如果是则调用doReleaseShared() 将 head 的状态从 -1 改为 0 并唤醒老二,这时 t3 在 doAcquireShared内 parkAndCheckInterrupt() 处恢复运行

    这回再来一次 for (;;) 执行 tryAcquireShared 成功则让读锁计数加一

    这时 t3 已经恢复运行,接下来 t3 调用 setHeadAndPropagate(node, 1),它原本所在节点被置为头节点

    下一个节点不是 shared 了,是写锁(和ReentrantLock一样,独占锁)因此不会继续唤醒 t4 所在节点

    2.4 t2 r.unlock,t3 r.unlock

    2.4.1 读锁释放流程与写锁加锁流程

    1、releaseShared(int arg)方法

    static final class NonfairSync extends Sync { // ReadLock 方法, 方便阅读, 放在此处 public void unlock() { sync.releaseShared(1); } // AQS 继承过来的方法, 方便阅读, 放在此处 public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }

    2、tryReleaseShared(int unused)方法

    t2 进入 sync.releaseShared(1) 中,调用 tryReleaseShared(1) 让计数减一,但由于计数还不为零

    // Sync 继承过来的方法, 方便阅读, 放在此处 protected final boolean tryReleaseShared(int unused) { // ... 省略不重要的代码 for (;;) { int c = getState(); int nextc = c - SHARED_UNIT; if (compareAndSetState(c, nextc)) { // 读锁的计数不会影响其它获取读锁线程, 但会影响其它获取写锁线程 // 计数为 0 才是真正释放 return nextc == 0; } } }

    3、doReleaseShared()方法

    t3 进入 sync.releaseShared(1) 中,调用 tryReleaseShared(1) 让计数减一,这回计数为零了,进入doReleaseShared() 将头节点从 -1 改为 0 并唤醒老二,即

    // AQS 继承过来的方法, 方便阅读, 放在此处 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; // 如果有其它线程也在释放读锁,那么需要将 waitStatus 先改为 0 // 防止 unparkSuccessor 被多次执行 if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases unparkSuccessor(h); } // 如果已经是 0 了,改为 -3,用来解决传播性,见后文信号量 bug 分析 else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; } } } 之后 t4 (写锁被唤醒)在 acquireQueued 中 parkAndCheckInterrupt 处恢复运行, 再次 for (;;) 这次自己是老二,并且没有其他竞争,tryAcquire(1) 成功, 修改头结点,流程结束

    3. ReentrantReadWriteLock锁降级与升级

    锁降级: 由上面的源码可以看出,线程在获取读锁时,如果state!=0,那么会先判断获取写锁的线程是不是当前线程,也就是说一个线程在获取写锁后,还可以获取读锁,当写锁释放后,就降级为读锁了。 不可以锁升级:

    Processed: 0.018, SQL: 9