Read-Write Lock读写锁模式的实现_Java

读写锁模式 read-write lock

read-write lock（读写锁模式）：允许多个线程同时读取共享资源，但在进行写入操作时必须独占资源，不再允许其他线程读或写。

⨳ 共享读（shared read）：读取操作可以同时被多个线程执行，因为读取操作不修改共享资源的状态，所以可以并行进行。

⨳ 独占写（exclusive write）：写入操作则必须在没有其他线程正在读取或写入时进行，以确保写入操作的原子性和一致性。

读写锁模式就是在读取操作和写入操作之间进行区分，想想也是，当线程“读取”实例的状态时，实例的状态不会发生变化。实例的状态仅在线程执行“写入”操作时才会发生变化。从实例的状态变化这个观点来看，“读取”和“写入”有着本质的区别。

基本实现

读写锁 readwritelock

读写锁该怎么实现呢？如果以守护条件的角度：

⨳ 读锁的获取：如有线程正在执行写入，则 wait 等待，也就是说，读锁获取的守护条件是无正在写入的线程。

⨳ 写锁的获取：如果有线程正在写入，则 wait 等待，如果有线程正在读取，也要wait 等待，也就是说，写锁获取的守护条件是无正在写入和读取的线程。

⨳ 读锁的释放：读操作执行完了即可释放，可以唤醒正在等待获取写锁的线程。

⨳ 写锁的释放：写操作执行完了即可释放，可以唤醒正在等待获取写锁和读锁的线程。

有了守护条件，就可以根据其设置成受保护资源，然后对其的操作加 synchronized 锁保护了：

public class readwritelock {

    private int readings = 0; // 正在读的线程数
    private int writings = 0; // 正在写的线程数

    // 获取读锁
    public synchronized void readlock() throws interruptedexception {
        while(writings>0){
            wait();
        }
        readings++;
    }

    // 写锁的获取
    public synchronized void writelock() throws interruptedexception {
        while (readings>0 || writings>0){
            wait();
        }
        writings++;
    }

    // 释放读锁
    public synchronized void readunlock(){
        readings--;
        notifyall();
    }

    // 释放写锁
    public synchronized void writeunlock(){
        writings--;
        notifyall();
    }
}

当然这个实现很简陋，比如读锁是不支持重入的，同一线程持有读锁后再调 writelock() 会自死锁（自己等自己释放）。

受保护对象 guardedobject

用读写锁代替排他锁保护受保护资源：

public class guardedobject {

    // 被守护的数字
    private int guardednum=0;

    private readwritelock lock = new readwritelock();

    // 读锁保护 对 guardednum 的读取
    public int getguardednum() throws interruptedexception {
        lock.readlock(); // 获取读锁
        try{
            thread.sleep(50); // 模拟读操作
            system.out.println(thread.currentthread().getname()+"正在读取："+guardednum);
            return guardednum;
        }finally {
            lock.readunlock(); // 释放读锁
        }
    }

    // 写锁保护 对 guardednum 的更改
    public int inc() throws interruptedexception {
        lock.writelock(); // 获取读锁
        try{
            thread.sleep(70); // 模拟读操作
            system.out.println(thread.currentthread().getname()+"正在更改："+guardednum);
            guardednum++;
            system.out.println(thread.currentthread().getname()+"更改后："+guardednum);
            return guardednum;
        }finally {
            lock.writeunlock(); // 释放读锁
        }
    }


}

注意，上述代码受保护资源 guardednum，由一把锁保护（readwritelock）。

如多线程同时调用 getguardednum 方法：

⨳ 获取读锁 lock.readlock() 操作是由排他锁 synchronized 保护，所以同一时刻只有一个线程进入readlock() 方法。

⨳ 获取读锁后续业务逻辑操作，不受任何任何排他锁 synchronized 保护，也就是说理论上，获取读锁后续操作允许多个线程同时执行，但这多个线程只有在执行完获取读锁操作后，才能继续执行，获取读锁会将不符合守护条件（无正在写入的线程）的线程挡在了外边。

如多线程同时调用 inc() 递增方法：

⨳ 获取写锁 lock.writelock() 操作同样由排他锁 synchronized 保护。

⨳ 获取写锁后续业务逻辑操作，也不受任何任何排他锁 synchronized 保护，也允许多个线程同时执行，只是获取写锁的线程如不符合写锁的保护条件（无正在写入和读取的线程），同样被阻塞在获取写锁的时刻。

注意，无论是使用读锁保护资源还是写锁保护资源，都需要合理的释放锁。

客户端 client

下面使用三个读线程，两个写线程同时访问 guardedobject 中的受保护资源：

    final random random = new random();
    final guardedobject guardedobject = new  guardedobject();
    new thread(()->{
        while (true){
            try {
                thread.sleep(random.nextint(3000));
                guardedobject.inc();
            } catch (interruptedexception e) {
                break;
            }
        }

    },"递增线程a").start();

    new thread(()->{
        while (true){
            try {
                thread.sleep(random.nextint(3000));
                guardedobject.inc();
            } catch (interruptedexception e) {
                break;
            }
        }

    },"递增线程b").start();

    new thread(()->{
        while (true){
            try {
                thread.sleep(random.nextint(3000));
                guardedobject.getguardednum();
            } catch (interruptedexception e) {
                break;
            }
        }

    },"读取线程a").start();

    new thread(()->{
        while (true){
            try {
                thread.sleep(random.nextint(3000));
                guardedobject.getguardednum();
            } catch (interruptedexception e) {
                break;
            }
        }

    },"读取线程b").start();

    new thread(()->{
        while (true){
            try {
                thread.sleep(random.nextint(3000));
                guardedobject.getguardednum();
            } catch (interruptedexception e) {
                break;
            }
        }

    },"读取线程c").start();
}

输出如下：

读取线程a正在读取：0
递增线程b正在更改：0
递增线程b更改后：1
递增线程a正在更改：1
递增线程a更改后：2
读取线程a正在读取：2
递增线程a正在更改：2
递增线程a更改后：3
读取线程c正在读取：3
递增线程a正在更改：3
递增线程a更改后：4
读取线程b正在读取：4
读取线程c正在读取：4
读取线程a正在读取：4
递增线程a正在更改：4
递增线程a更改后：5
...

可以看到用读写锁代替排它锁保护 guardednum 时，guardednum 也不会出现并发问题。

可重入读写锁

juc包下就有现成读写锁实现 —— reentrantreadwritelock

public class reentrantreadwritelock
    /** inner class providing readlock */
    private final reentrantreadwritelock.readlock readerlock;
    /** inner class providing writelock */
    private final reentrantreadwritelock.writelock writerlock;
    final sync sync;
    
   public reentrantreadwritelock(boolean fair) {
    sync = fair ? new fairsync() : new nonfairsync();
    readerlock = new readlock(this);
    writerlock = new writelock(this);
  }

和可重入锁 reentrantlock 类似，核心功能也是静态内部类实现的。

reentrantlock 是内部有个 sync 实现 aqs，有个 fairsync 和 nonfairsync 分别实现 aqs，重新 tryacquire 方法，实现公平锁和非公平锁。

reentrantreadwritelock 也一样，也有一个实现 aqs 的内部类 sync，也有 fairsync 和 nonfairsync 。

aqs 上节刚讲完，里面有个 volatile 修饰的成员变量 state，记录线程占用状态（state = 0 表示锁空闲）与可重入次数，aqs 的父类 aos 还记录着当前占用的线程引用。

public abstract class abstractqueuedsynchronizer
    extends abstractownablesynchronizer
    implements java.io.serializable {
    
    private transient volatile node head;
    private transient volatile node tail;
    private volatile int state;
    
 
public abstract class abstractownablesynchronizer
    implements java.io.serializable {
    private transient thread exclusiveownerthread;

aqs 就一个 state，怎么记录两个锁（读锁和写锁）的线程占用状态与可重入次数呢？

reentrantreadwritelock 很聪明， state 是 int 类型的变量，int 一共 32 位，能存储上亿的数字，那就让高 16 位存放读锁重入总次数，低 16 位 记录写锁重入次数，这样拆分，还能记录 65,535 次重入次数，谁家好线程能重入这么多。

读写锁线程占用状态和对重入次数的问题解决，那占用的线程记录在哪呢？aos 就一个 exclusiveownerthread 肯定是不够的。

所以对于写锁来说，同一时刻最多只有一个写线程持有，所以直接复用 aqs 的 exclusiveownerthread 就够了。

对于读锁来说，就要 reentrantreadwritelock 的内部类 sync 自己搞一个地方存读线程的信息了。

下面结合代码看看具体是怎么实现的。

抽象队列同步器 aqs

abstract static class sync extends abstractqueuedsynchronizer {

    /**
     * state 的位分割参数。
     * aqs 只有一个 int state，读写锁需要同时记录两类计数，
     * 因此约定：高 16 位 = 读锁（共享）重入总次数，低 16 位 = 写锁（独占）重入次数。
     */
    static final int shared_shift   = 16;
    /** 共享计数每次增减的步长，即 1 << 16，用于对高 16 位做加减 */
    static final int shared_unit    = (1 << shared_shift);
    /** 读锁计数上限，65535；超出说明使用方式异常，直接抛 error */
    static final int max_count      = (1 << shared_shift) - 1;
    /** 独占（写锁）计数的掩码，用于从 state 中剥离出低 16 位 */
    static final int exclusive_mask = (1 << shared_shift) - 1;

    /**
     * 单个线程的读锁持有计数。
     * tid 在构造时即固化，用于后续校验缓存是否属于当前线程，
     * 避免线程复用（线程池场景）导致的误命中。
     */
    static final class holdcounter {
        int count;          // initially 0
        final long tid = locksupport.getthreadid(thread.currentthread());
    }

    /**
     * 读锁计数的兜底存储：每个线程一份独立的 holdcounter。
     * 之所以用 threadlocal，是因为读锁是共享的，可能有 n 个线程同时持有，
     * aqs 的 exclusiveownerthread 只能标识一个独占线程，无法覆盖读线程。
     */
    static final class threadlocalholdcounter
        extends threadlocal<holdcounter> {
        public holdcounter initialvalue() {
            return new holdcounter();
        }
    }

    /**
     * 缓存"上一次释放读锁"的线程的 holdcounter。
     * 嵌套读释放时，同一线程往往连续操作，可直接命中此字段，
     * 省掉一次 threadlocal 查询。属于纯性能优化，不参与正确性判断。
     */
    private transient holdcounter cachedholdcounter;

    /**
     * 记录第一个获取读锁的线程。
     * 针对"系统中只有一个读者"这一最高频场景做零开销优化。
     */
    private transient thread firstreader;
    /** 与 firstreader 配对，记录其读锁重入次数 */
    private transient int firstreaderholdcount;

    sync() {
        readholds = new threadlocalholdcounter();
        // 空写一次 state，借助 volatile 写语义建立 happens-before，
        // 保证其他线程能看到 readholds 的初始化结果
        setstate(getstate());
    }
}

代码还是很清晰的，holdcounter 同时承担了"身份标识"和"计数"两个职责，既记录线程的id，又记录重入次数。

这里用 locksupport.getthreadid(current) 而不是直接比对象引用，是为了在线程池复用场景下避免缓存误命中。

每个读线程自己的存储柜 threadlocal 记录自己的 holdcounter。

而且对于读线程状态的查询，还有三层缓存的完整查找链

final int getreadholdcount() {
    if (getreadlockcount() == 0)
        return 0;

    thread current = thread.currentthread();
    // 第1层：firstreader 是否就是当前线程？
    // 直接返回 firstreaderholdcount（零开销，无 threadlocal 查询）
    if (firstreader == current)
        return firstreaderholdcount;
    // cachedholdcounter 的 tid 是否匹配当前线程？
    // 直接返回 count  
    holdcounter rh = cachedholdcounter;
    if (rh != null && rh.tid == locksupport.getthreadid(current))
        return rh.count;
    // 第3层：threadlocal 查询（兜底，覆盖所有其他线程）
    int count = readholds.get().count;
    if (count == 0) readholds.remove();
    return count;
}

下面就结合读锁、写锁具体获取锁，释放锁的方法，看看这三层缓存是怎么协作的？

读锁 readlock

public static class readlock implements lock, java.io.serializable {
    private final sync sync;
    protected readlock(reentrantreadwritelock lock) {
        sync = lock.sync;
    }
    public void lock() {
        sync.acquireshared(1);
    }
    public void unlock() {
        sync.releaseshared(1);
    }

读锁的 lock 方法是委派 aqs 的 acquireshared 方法实现的。

  public final void acquireshared(int arg) {
    if (tryacquireshared(arg) < 0)
        acquire(null, arg, true, false, false, 0l);
  }

aqs 的子类也就是 reentrantreadwritelock 的内部类 sync 实现了 tryacquireshared 方法，标准的模板方法。

protected final int tryacquireshared(int unused) {
 
    thread current = thread.currentthread();
    int c = getstate();

    // ① 写锁互斥检查：低 16 位非 0 说明有写锁被持有；
    //    若持有者不是当前线程，读锁直接获取失败（返回 -1 表示未获取到）。
    //    若持有者正是当前线程，则属于"写锁降级为读锁"的合法路径，继续往下走。
    if (exclusivecount(c) != 0 &&
        getexclusiveownerthread() != current)
        return -1;

    // ② 取出高 16 位，即当前读锁的总重入次数
    int r = sharedcount(c);

    // ③  cas 成功：把高 16 位加一个 shared_unit，原子地增加读锁计数。
    if (compareandsetstate(c, c + shared_unit)) {

        // ④ state 更新成功后，再维护"每个线程的读锁持有计数"，分三种情况：
        if (r == 0) {
            // 之前没有任何读线程，当前线程就是第一个读者，
            // 用两个普通字段直接记录，避免走 threadlocal
            firstreader = current;
            firstreaderholdcount = 1;
        } else if (firstreader == current) {
            // 当前线程就是第一个读者，属于其重入，直接自增字段计数
            firstreaderholdcount++;
        } else {
            // 其他线程：走缓存 + threadlocal 兜底
            holdcounter rh = cachedholdcounter;
            if (rh == null ||
                rh.tid != locksupport.getthreadid(current)) {
                // 缓存为空或 tid 不匹配（可能是线程池复用导致），
                // 从 threadlocal 取当前线程自己的 holdcounter，并刷新缓存
                cachedholdcounter = rh = readholds.get();
            } else if (rh.count == 0) {
                // 缓存命中但计数为 0，说明之前被清理过，需要重新放回 threadlocal
                readholds.set(rh);
            }
            rh.count++;
        }
        return 1;   // 返回 >= 0 表示获取成功，aqs 会据此决定是否传播唤醒
    }

    // ⑤ 快速路径没走通
    //    交给带完整自旋重试逻辑的 full 版本处理，其中也包含重入场景
    return fulltryacquireshared(current);
}

代码还是很清晰的，线程进入读锁的前提是没有其他线程的写锁；或者虽有写锁，但调用线程和持有写锁的线程是同一个。

因为读与读之间不互斥，所以不需要检查"有没有其他读锁"。

如果获取读锁时，有线程持有写锁，tryacquireshared 会返回 -1，读线程依旧会加入阻塞队列，如果只是其他线程已抢先获取到了读锁，那会 cas 失败，到 fulltryacquireshared 方法中自旋再次获取读锁。

读锁的释放同样是委派 aqs 的 releaseshared 方法实现的。

public final boolean releaseshared(int arg) {
    if (tryreleaseshared(arg)) {
        signalnext(head);
        return true;
    }
    return false;
}

同样，tryreleaseshared 方法的实现还是由reentrantreadwritelock 的内部类 sync 完成。

protected final boolean tryreleaseshared(int unused) {
    thread current = thread.currentthread();

    // ① 先维护"当前线程自己的读锁重入计数"，再动全局 state，
    //    顺序不能反：线程级计数是释放合法性的前置校验
    if (firstreader == current) {
        // 当前线程是第一个读者，走零开销的普通字段路径，不碰 threadlocal
        // assert firstreaderholdcount > 0;
        if (firstreaderholdcount == 1)
            firstreader = null;      // 计数归零，第一个读者的身份随之清除
        else
            firstreaderholdcount--;  // 仍有重入，仅递减计数
    } else {
        // ② 非 firstreader：先尝试命中缓存，未命中再走 threadlocal 兜底
        holdcounter rh = cachedholdcounter;
        if (rh == null ||
            rh.tid != locksupport.getthreadid(current))
            // 缓存为空或 tid 不匹配（线程池复用场景），取当前线程自己的计数器
            rh = readholds.get();

        int count = rh.count;
        if (count <= 1) {
            // 本次释放后该线程的读锁计数将归零，直接从 threadlocal 移除，
            // 避免线程池场景下 entry 长期驻留造成内存泄漏
            readholds.remove();
            if (count <= 0)
                // count 为 0 说明当前线程并未持有读锁，属于未配对解锁
                throw unmatchedunlockexception();
        }
        --rh.count;
    }

    // ③ 全局计数更新：自旋 cas，把 state 高 16 位减一个 shared_unit
    for (;;) {
        int c = getstate();
        int nextc = c - shared_unit;
        if (compareandsetstate(c, nextc))
            // releasing the read lock has no effect on readers,
            // but it may allow waiting writers to proceed if
            // both read and write locks are now free.
            // 释放读锁对其它读者没有影响，但若此时读写锁都已空闲，
            // 则可能让等待中的写线程得以继续
            return nextc == 0;
    }
}

解锁是加锁的相反操作，其实从这个代码中也能看出来 aqs 中的 state 高 16 位记录的不是某一个读线程的重入次数，是所有读线程的重入次数。

写锁 writelock

public static class writelock implements lock, java.io.serializable {
    private final sync sync;

    protected writelock(reentrantreadwritelock lock) {
        sync = lock.sync;
    }
    public void lock() {
        sync.acquire(1);
    }
    public void unlock() {
        sync.release(1);
    }

写锁的加锁和释放就简单了，走的是独占模式，不需要像读锁那样维护 per-thread 的计数体系。

和 reentrantlock 一样，都是通过模板方法模式，调用自己内部类 sync，直接看模板方法tryacquire：

protected final boolean tryacquire(int acquires) {
 
    thread current = thread.currentthread();
    int c = getstate();
    // 取出低 16 位，即当前写锁的重入计数
    int w = exclusivecount(c);

    if (c != 0) {
        // (note: if c != 0 and w == 0 then shared count != 0)
        // 注意：c 非零但 w 为零，说明高 16 位有值，即当前有读锁被持有
        if (w == 0 || current != getexclusiveownerthread())
            // 两种失败情形：
            // ① w == 0：有读锁存在，写锁不能获取（不支持锁升级，避免多读线程同时升级死锁）
            // ② 写锁已被其他线程持有：独占语义下直接失败
            return false;

        // 走到这里说明：写锁持有者就是当前线程，属于重入场景
        if (w + exclusivecount(acquires) > max_count)
            // 重入次数超出低 16 位上限，属于使用异常
            throw new error("maximum lock count exceeded");

        // reentrant acquire
        // 重入时直接 setstate 即可，无需 cas：
        // 独占语义下已确认持有者是自己，不存在其他线程并发修改低 16 位
        setstate(c + acquires);
        return true;
    }

    // c == 0：读写锁都空闲，尝试首次获取
    if (writershouldblock() ||
        !compareandsetstate(c, c + acquires))
        // 两个失败原因：
        // ① writershouldblock()：按队列策略当前写线程需要排队
        //    （公平实现用 hasqueuedpredecessors() 判断队列中是否有前驱；非公平实现直接返回 false）
        // ② cas 失败：并发下有别的线程抢先修改了 state
        return false;

    // 获取成功，登记独占持有者
    setexclusiveownerthread(current);
    return true;
}