【多线程】读写锁ReentrantReadWriteLock源码分析

• （一）读写锁ReentrantReadWriteLock源码分析
• • 【1】类图结构和参数
  • 【2】写锁的获取和释放
  • • （1）lock方法
    • （2）lockInterruptibly方法
    • （3）tryLock方法
    • （4）unlock方法
  • 【3】读锁的获取和释放
  • • （1）lock方法
    • （2）lockInterruptibly方法
    • （3）tryLock方法
    • （4）unlock方法
• （二）读写锁ReentrantReadWriteLock使用案例
• • （1）案例一：对缓存token的读写
  • （2）案例二：测试

ReentrantLock完全可以解决线程安全问题，但是ReentrantLock是独占锁，某一时间只能有一个线程可以获取锁，其他线程只能等待。现实中有很多“写少读多”的场景，那么我们就希望同一时间只能有一个写线程可以获取写锁，但是同时可以多个读线程可以获取读锁，于是便有了ReentrantReadWriteLock。ReentrantReadWriteLock采用读写分离策略，允许多个线程可以同时获取读锁。

（一）读写锁ReentrantReadWriteLock源码分析

【1】类图结构和参数

读写锁的内部维护了一个读锁ReadLock和一个写锁WriteLock，它们依赖Sync实现具体的功能。而Sync继承自AQS，并且也提供了公平锁和非公平锁的实现。AQS中只维护了一个state状态，而ReentrantReadWriteLock需要维护读状态和写状态，一个state怎么表示写和读两种状态呢？其实就是使用state的高1位表示读状态，获取到读锁的次数；使用低16位表示获取到写锁的线程的可重入次数。

ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
Lock readLock = lock.readLock();
Lock writeLock = lock.writeLock();

【2】写锁的获取和释放

（1）lock方法

写锁是个独占锁，某时只有一个线程可以获取锁，如果当前没有线程获取到读锁和写锁，则当前线程可以获取到写锁然后返回。如果当前有线程获取到读锁和写锁，则当前请求写锁的线程会被阻塞挂起。

另外，写锁是可重入锁，如果当前线程已经获取了该锁，再次获取只是简单地把可重入次数+1后直接返回。

【1】tryAcquire方法

/*** 这个方法是写锁调用才会执行的* 不是第一次加写锁* 因为是写锁才调用的方法，因此只需要排除前面加的都是读锁这种情况即可，也就是 c!=0 但 w==0的情况* 先判断是否加了锁c！=0，如果加了锁也就是c!=0内部的这个分支，* 是否没有加过写锁* 是否是重入写锁* 是第一次加写锁*/
@ReservedStackAccess
protected final boolean tryAcquire(int acquires) {Thread current = Thread.currentThread();// 获取当前线程对象int c = getState(); // 获取state（前16位是读锁个数、后16位是写锁个数）int w = exclusiveCount(c);// 获得写锁的个数，w有write的含义。这个值就是写锁的个数，通过按位与 15 得到写锁个数if (c != 0) {// c！=0则说明加过锁// 如果写锁个数为0 （说明加的都是读锁，不需要阻塞因此抢占锁失败） 或者 当前线程不是持有写锁线程(w!=0说明加过写锁需要判断当前线程是否是持有写锁的那个线程，不是则说明抢占锁失败)if (w == 0 || current != getExclusiveOwnerThread())return false; // 表示抢占锁失败，这里导致了两种情况，一种是加的都是读锁，一种是加了写锁，但当前线程不是持有锁线程// 执行下面的判断都表示加过了写锁，相当于写锁的重入，因此需要将写锁计数相加也就是判断里的操作if (w + exclusiveCount(acquires) > MAX_COUNT) // 说明是重入锁，判断本次加了acquires次锁后锁计数是否超过最大值 2的16次方-1throw new Error("Maximum lock count exceeded");// 超过能加写锁的最大值则抛异常// 写锁重入，因此保留读锁加上写锁重入的acquires次，将state更新setState(c + acquires);return true;//返回true说明加锁成功}// 前面没有加过锁，需要加写锁，尝试利用CAS操作更新state进行加锁，实际上逻辑上不需要这里的if，但是应该是由于并发问题怕中途state值被改了，因此CAS操作可能失败（所以失败则return false）// c==0 说明没有加过锁，尝试将state从0更新为acquires，更新成功则说明加锁成功，因此不会返回false，而是执行后面的return trueif (writerShouldBlock() || !compareAndSetState(c, c + acquires))return false;setExclusiveOwnerThread(current);// 将当前线程设置为独占线程，表示加写锁成功！return true;// 加锁成功
}

【2】exclusiveCount方法
直接将状态state和（2^16 - 1）做与运算，其等效于将state模上2^16。写锁数量由state的低十六位表示。

static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
/*** 返回写锁的个数*/
static int exclusiveCount(int c) {return c & EXCLUSIVE_MASK;
}

【3】writerShouldBlock方法
FairSync公平锁的写法

public final boolean hasQueuedPredecessors() {// The correctness of this depends on head being initialized// before tail and on head.next being accurate if the current// thread is first in queue.Node t = tail; // Read fields in reverse initialization orderNode h = head;Node s;return h != t &&((s = h.next) == null || s.thread != Thread.currentThread());
}static final class FairSync extends Sync {private static final long serialVersionUID = -2274990926593161451L;final boolean writerShouldBlock() {return hasQueuedPredecessors();}final boolean readerShouldBlock() {return hasQueuedPredecessors();}
}

NonfairSync非公平锁写法

final boolean apparentlyFirstQueuedIsExclusive() {Node h, s;return (h = head) != null &&(s = h.next)  != null &&!s.isShared()         &&s.thread != null;
}static final class NonfairSync extends Sync {private static final long serialVersionUID = -8159625535654395037L;final boolean writerShouldBlock() {return false; // writers can always barge}final boolean readerShouldBlock() {/* As a heuristic to avoid indefinite writer starvation,* block if the thread that momentarily appears to be head* of queue, if one exists, is a waiting writer.  This is* only a probabilistic effect since a new reader will not* block if there is a waiting writer behind other enabled* readers that have not yet drained from the queue.*/return apparentlyFirstQueuedIsExclusive();}
}

【4】获取写锁的步骤
（1）首先获取c、w。c表示当前锁状态；w表示写线程数量。然后判断同步状态state是否为0。如果state!=0，说明已经有其他线程获取了读锁或写锁，执行(2)；否则执行(5)。

（2）如果锁状态不为零（c != 0），而写锁的状态为0（w = 0），说明读锁此时被其他线程占用，所以当前线程不能获取写锁，自然返回false。或者锁状态不为零，而写锁的状态也不为0，但是获取写锁的线程不是当前线程，则当前线程也不能获取写锁。

（3）判断当前线程获取写锁是否超过最大次数，若超过，抛异常，反之更新同步状态（此时当前线程已获取写锁，更新是线程安全的），返回true。

（4）如果state为0，此时读锁或写锁都没有被获取，判断是否需要阻塞（公平和非公平方式实现不同），在非公平策略下总是不会被阻塞，在公平策略下会进行判断（判断同步队列中是否有等待时间更长的线程，若存在，则需要被阻塞，否则，无需阻塞），如果不需要阻塞，则CAS更新同步状态，若CAS成功则返回true，失败则说明锁被别的线程抢去了，返回false。如果需要阻塞则也返回false。

（5）成功获取写锁后，将当前线程设置为占有写锁的线程，返回true。

（2）lockInterruptibly方法

（3）tryLock方法

/*** 执行tryLock进行写入，从而在两种模式下都可以进行插入。 这与tryAcquire的作用相同，只是缺少对writerShouldBlock的调用。*/
@ReservedStackAccess
final boolean tryWriteLock() {Thread current = Thread.currentThread(); // 得到当前线程int c = getState(); // 得到锁计数if (c != 0) { // 不为0说明加过锁int w = exclusiveCount(c); // 得到写锁次数if (w == 0 || current != getExclusiveOwnerThread())return false;// 写锁被其它线程占用，当前线程抢占写锁失败if (w == MAX_COUNT)throw new Error("Maximum lock count exceeded");}// 第一次就加写锁，cas更新state值if (!compareAndSetState(c, c + 1))return false;// 将当前线程设置为独占setExclusiveOwnerThread(current);return true;// 写锁加锁成功！
}

（4）unlock方法

/*** 尝试释放锁*/
@ReservedStackAccess
protected final boolean tryRelease(int releases) {if (!isHeldExclusively())//判断当前现是否是持有锁线程如果是则不执行，如果不是则需要抛异常，因为当前线程没有持有锁throw new IllegalMonitorStateException();int nextc = getState() - releases;// 计算释放锁释放合法，不允许释放超过加锁次数boolean free = exclusiveCount(nextc) == 0;//if (free) // 判断释放锁后释放锁计数是否为0，为0则说明当前线程不再是持有锁线程将其从排他线程状态清除setExclusiveOwnerThread(null);setState(nextc);// 更新锁次数计数return free;//如果释放锁后计数为0则返回true，否则返回false
}

写锁的释放过程还是相对而言比较简单的：首先查看当前线程是否为写锁的持有者，如果不是抛出异常。然后检查释放后写锁的线程数是否为0，如果为0则表示写锁空闲了，释放锁资源将锁的持有线程设置为null，否则释放仅仅只是一次重入锁而已，并不能将写锁的线程清空。

说明：此方法用于释放写锁资源，首先会判断该线程是否为独占线程，若不为独占线程，则抛出异常，否则，计算释放资源后的写锁的数量，若为0，表示成功释放，资源不将被占用，否则，表示资源还被占用。其方法流程图如下。

【3】读锁的获取和释放

（1）lock方法

public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0)doAcquireShared(arg);
}

【1】tryAcquireShared方法

/*** 读锁才调用的方法，当前线程尝试获取读锁*/
@ReservedStackAccess
protected final int tryAcquireShared(int unused) {Thread current = Thread.currentThread(); // 获取当前线程int c = getState();// 获取存有读和写锁次数的state值/*** 是写锁则进入*/// 通过exclusiveCount(c)得到写锁次数，如果不为0则说明加了写锁。加了写锁需要判断当前线程是否是持有写锁的线程，是则不返回-1，不是则说明是写读状态需要进行阻塞当前线程if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)return -1; // 说明是写读状态、返回-1，抢占读锁失败// 执行到这里说明前面没有加过写锁，可能加过读锁int r = sharedCount(c); // 获取加的读锁次数，r就是read，实际就是将state右移16位得到// 到这里说明没有加过锁，到这里c是0，因此进行加锁操作将state更新为读锁的1 实际二进制是：0000 0000 0000 0001 0000 0000 0000 0000/***  是读锁，*  一、读是共享的情况直接执行if内*/if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) {if (r == 0) { // 第一次进入，因为能到达这里就说明没有写锁，有判断r==0则说明读锁也为0，则说明是第一次调用firstReader = current; // 将第一个线程存起来firstReaderHoldCount = 1;// 计数为1} else if (firstReader == current) {firstReaderHoldCount++; // 读重入，读锁计数进行累加} else {// 说明不是获得读锁的线程进来了// tid 为key ，value为读锁次数HoldCounter rh = cachedHoldCounter;// 将当前线程初始值是null// 第一次null直接创建一个if (rh == null || rh.tid != LockSupport.getThreadId(current))cachedHoldCounter = rh = readHolds.get();// 通过ThreadLocal得到HoldCounter（计数保持器，内部存了加锁计数）else if (rh.count == 0) // 如果锁计数为0readHolds.set(rh); // 更新锁计数保持器对象rh.count++; // 计数累加}return 1;// 表示抢占读锁成功}/*** 二、读是排他的情况，调用下面这个方法*/return fullTryAcquireShared(current);
}

【2】doAcquireShared方法

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) {int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCif (interrupted)selfInterrupt();failed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}
}

【3】sharedCount方法
表示占有读锁的线程数量，直接将state右移16位，就可以得到读锁的线程数量，因为state的高16位表示读锁，对应的第十六位表示写锁数量。

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

【4】fullTryAcquireShared方法

/*** 读是排他的情况采用自旋方式* 完整版本的获取读，可处理CAS错误和tryAcquireShared中未处理的可重入读。*/
final int fullTryAcquireShared(Thread current) {/*** 该代码与tryAcquireShared中的代码部分冗余，但由于不使tryAcquireShared与重试和延迟读取保持计数之间的交互复杂化，因此整体代码更简单。*/HoldCounter rh = null;for (; ; ) {// 自旋int c = getState(); // 获取读写锁计数/*** 如果存在写锁*/if (exclusiveCount(c) != 0) {if (getExclusiveOwnerThread() != current)// 判断当前线程是否是持有同一把写锁的线程return -1;// 加锁失败，当前线程不是持有写锁线程}/*** 不存在写的情况*/// 1.判断读是否是排他的，如果是则进入else if (readerShouldBlock()) {// 当前线程是不是第一个读锁线程，是则说明当前线程是重入的读锁线程if (firstReader == current) {// 什么也没有} else {// 如果当前线程不是第一个抢占到读锁的线程，如果锁计数存在if (rh == null) {rh = cachedHoldCounter;  // 得到锁计数保持器if (rh == null || rh.tid != LockSupport.getThreadId(current)) {rh = readHolds.get(); // 得到锁计数保持器if (rh.count == 0) // 如果计数为0readHolds.remove(); // 清除保持器}}// 读锁计数保持器存在，如果等于0则抢占读锁失败，因为这个计数器在tryAcquireShared方法已经被赋值了，所以不会为0，为0说明cas操作失败了if (rh.count == 0)return -1; // 加锁失败，当前线程}}// 2.到这里说明是共享的读/*** 注意：*  如果是tryAcquireShared方法过来的其实下面不会执行到的，*  因为在tryAcquireShared方法中已经走过一遍这个逻辑了，*  这里加上这个逻辑只是处于对当前方法的封装，这样当前方法可以不用依赖tryAcquireShared方法*/if (sharedCount(c) == MAX_COUNT) // 判断读锁是否超过最大值throw new Error("Maximum lock count exceeded");// 读共享，因此只需要通过cas将读锁计数累加1即可，因为CAS操作多以是单线程所以是加1if (compareAndSetState(c, c + SHARED_UNIT)) {// 更新state值// c 一开始是0，因为上面更新的不是c而是state值，如果c是0说明是第一个线程调用了这个方法，执行到了这里if (sharedCount(c) == 0) {firstReader = current; // 保存当前的第一个线程firstReaderHoldCount = 1;// 保存计数（因为是第一次进入所以是1）} else if (firstReader == current) {firstReaderHoldCount++; // 持锁的同一个线程重入读锁} else {if (rh == null)rh = cachedHoldCounter; // 其它线程尝试获取读锁，获取第一个线程产生的HoldCounter对象if (rh == null || rh.tid != LockSupport.getThreadId(current))rh = readHolds.get(); // 从ThreadLocal中获取HoldCounter对象else if (rh.count == 0)readHolds.set(rh); // 如果锁计数为0更新锁计数保持其对象rh.count++; // 读锁计数累加cachedHoldCounter = rh; // 保存读锁计数器对象}return 1; // 读锁加锁成功}}
}

【4】获取读锁的流程
读锁获取锁的过程比写锁稍微复杂些，首先判断写锁是否为0并且当前线程不占有独占锁，直接返回；否则，判断读线程是否需要被阻塞并且读锁数量是否小于最大值并且比较设置状态成功，若当前没有读锁，则设置第一个读线程firstReader和firstReaderHoldCount；若当前线程线程为第一个读线程，则增加firstReaderHoldCount；否则，将设置当前线程对应的HoldCounter对象的值。流程图如下。

（2）lockInterruptibly方法

（3）tryLock方法

（4）unlock方法

【1】tryReleaseShared方法
此方法表示读锁线程释放锁。首先判断当前线程是否为第一个读线程firstReader，若是，则判断第一个读线程占有的资源数firstReaderHoldCount是否为1，若是，则设置第一个读线程firstReader为空，否则，将第一个读线程占有的资源数firstReaderHoldCount减1；若当前线程不是第一个读线程，那么首先会获取缓存计数器（上一个读锁线程对应的计数器 ），若计数器为空或者tid不等于当前线程的tid值，则获取当前线程的计数器，如果计数器的计数count小于等于1，则移除当前线程对应的计数器，如果计数器的计数count小于等于0，则抛出异常，之后再减少计数即可。无论何种情况，都会进入无限循环，该循环可以确保成功设置状态state。

@ReservedStackAccess
protected final boolean tryReleaseShared(int unused) {Thread current = Thread.currentThread();// 获取当前线程对象if (firstReader == current) { // 当前线程是否是第一个持有锁线程if (firstReaderHoldCount == 1) // 是否是第一次上锁后就解锁了firstReader = null; // 清除第一个读锁线程elsefirstReaderHoldCount--;// 将读锁计数减一} else {HoldCounter rh = cachedHoldCounter; // 得到缓存的计数器对象if (rh == null || rh.tid != LockSupport.getThreadId(current))rh = readHolds.get(); // 如果缓存的计数器对象不是当前线程的，则获取当前线程的计数器对象，重新赋值int count = rh.count; // 得到当前线程的读锁计数if (count <= 1) { // 释放锁后为0，或者过度释放，则移除计数器readHolds.remove();// 移除计数器if (count <= 0)throw unmatchedUnlockException();}--rh.count; // 锁计数减1}for (; ; ) {int c = getState(); // 获得锁计数int nextc = c - SHARED_UNIT; // 读锁计数减一if (compareAndSetState(c, nextc)) // cas操作更新state值// 释放读取锁定对读取器没有影响.但是,如果现在读和写锁都已释放,则可能允许等待的编写器继续进行.return nextc == 0;}
}

流程图如下：

在读锁的获取、释放过程中，总是会有一个对象存在着，同时该对象在获取线程获取读锁是+1，释放读锁时-1，该对象就是HoldCounter。

要明白HoldCounter就要先明白读锁。前面提过读锁的内在实现机制就是共享锁，对于共享锁其实我们可以稍微的认为它不是一个锁的概念，它更加像一个计数器的概念。一次共享锁操作就相当于一次计数器的操作，获取共享锁计数器+1，释放共享锁计数器-1。只有当线程获取共享锁后才能对共享锁进行释放、重入操作。所以HoldCounter的作用就是当前线程持有共享锁的数量，这个数量必须要与线程绑定在一起，否则操作其他线程锁就会抛出异常。

（二）读写锁ReentrantReadWriteLock使用案例

（1）案例一：对缓存token的读写

第三方接口调用需要带上访问凭据accessToken ，需要先调用login接口传入username password获取该token。每个业务请求之前都login获取新的accessToken ，因为accessToken 不是一次性的，是有有效时间的。所以使用 ReadWriteLock 对工具类的进行了改造。

减少调用login接口的次数，无须每次调用业务接口都获取一次token；那么Token就得缓存起来，进而不同线程独读写共享变量就有并发问题：例如，读取缓存token时，无需加锁，各个线程之间的读是并发的互不影响；但是当token失效时，就必须重新申请，这时就得刷新token的缓存，就涉及到了并发读写。所以，用到了ReadWriteLock。最终，线程之间并发读不受影响，但是同一时刻只有一个线程可以写，而且写的时候 不会有线程能读到token。还需要注意一个细节，有可能有多个线程同时触发了申请token的条件，但最终只会有一个会成功申请token.

package com.example.demo;import java.util.Date;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;/*** @author wang * @date 2022/9/23 10:37*/
public class ReadWriteLockDemo {private static final ReentrantReadWriteLock READ_WRITE_LOCK = new ReentrantReadWriteLock();private static final Lock READ_LOCK = READ_WRITE_LOCK.readLock();private static final Lock WRITE_LOCK = READ_WRITE_LOCK.writeLock();//初始化Token; 这里一般是spring 容器舒适化过程中，先发一个请求获取tokenprivate AccessToken accessToken = new AccessToken(UUID.randomUUID().toString(), new Date());static   int i = 0;private void login(AccessToken expireToken) {WRITE_LOCK.lock();try {//如果this.accessToken 的值依然是 过期请求的值，才重新申请if (expireToken.equals(this.accessToken)) {//模拟http请求 获取 accessTokenSystem.out.println(Thread.currentThread().getName() + "  申请了新的Token");//可以观察到 token申请期间 不会有doGet return相关数据Thread.sleep(1000);this.accessToken = new AccessToken(UUID.randomUUID().toString(), new Date());} else {System.out.println(Thread.currentThread().getName() + "其他线程已经更新过accessToken , 跳过申请token ");}} catch (InterruptedException e) {throw new RuntimeException(e);} finally {WRITE_LOCK.unlock();}}/*** 业务请求，需要用到 Token*/public void doGet() throws InterruptedException {boolean isExpire;READ_LOCK.lock();try {//携带token 发起业务请求；由于不关注业务请求的返回值，只关注请求返回中 是否提示token失效；下面这行就代表发起业务请求isExpire = accessToken.isExpire();//如果token失效，需要重新登录} finally {READ_LOCK.unlock();}if (isExpire) {System.out.println(Thread.currentThread().getName() + "  token 过期 重新申请");login(this.accessToken);//重新发送一次业务请求doGet();} else {System.out.println(Thread.currentThread().getName() + "  业务请求成功！ return 相关数据");i++;}}public static void main(String[] args) throws InterruptedException {ReadWriteLockDemo readWriteLockDemo = new ReadWriteLockDemo();for (int i = 0; i < 100; i++) {new Thread(() -> {try {readWriteLockDemo.doGet();Thread.sleep(new Random().nextInt(3));} catch (InterruptedException e) {throw new RuntimeException(e);}}).start();Thread.sleep(500);}System.out.println(i);}
}

package com.example.demo;import java.time.LocalDateTime;
import java.util.Date;/*** @author wang * @date 2022/9/23 11:14*/public class AccessToken {public AccessToken(String value, Date createTime) {this.value = value;this.createTime = createTime;}private String value;private Date createTime;public boolean isExpire() {try {//模拟业务请求响应时间Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}//假设token有效期是 2 秒;  一般情况下，是发送业务请求之后 返回提示 token是否失效；return (System.currentTimeMillis() / 1000) - (createTime.getTime() / 1000) > 2;}public String getValue() {return value;}public void setValue(String value) {this.value = value;}
}

（2）案例二：测试

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;public class T01_ReadWriteLock {static ReadWriteLock readWriteLock = new ReentrantReadWriteLock();static Lock readLock = readWriteLock.readLock();static Lock writeLock = readWriteLock.writeLock();public void read(Lock lock) throws InterruptedException {lock.lock();Thread.sleep(1000);System.out.println("read");lock.unlock();}public void write(Lock lock) throws InterruptedException {lock.lock();Thread.sleep(1000);System.out.println("write");lock.unlock();}public static void main(String[] args) {T01_ReadWriteLock t = new T01_ReadWriteLock();Runnable read = ()-> {try {t.read(readLock);} catch (InterruptedException e) {e.printStackTrace();}};Runnable write = ()-> {try {t.write(writeLock);} catch (InterruptedException e) {e.printStackTrace();}};for(int i = 0;i<20;i++) {new Thread(read).start();}for (int i = 0;i<3;i++) {new Thread(write).start();}}}