读写锁ReentrantReadWriteLock源代码浅析

1.简介

并发中常用的ReentrantLock，是一种典型的排他锁，这类锁在同一时刻只允许一个线程进行访问，实际上将并行操作变成了串行操作。在并发量大的业务中，其整体效率、吞吐量不能满足实现的需要。而且实际的业务中一般情况是读多于写，多个线程读操作不会改变已经有的数据，不会有数据的一致性问题，而一个写操作就会改变数据，其他的的读操作就可能读到过期的数据。读写锁正是为了这种业务需求而产生的，读写锁在同一时刻可以允许多个读线程访问，但是在写线程访问时，所有的读线程和其他写线程均被阻塞。读写锁维护了一对锁，一个读锁和一个写锁，通过分离读锁和写锁，使得并发性相比一般的排他锁有了很大提升。

读写锁同时支持公平锁和非公平锁，默认实现是非公平锁，非公平锁的吞吐量高于公平锁。

读写锁也支持重入，读锁和写锁的最大重入次数是65535次，这是由int类型本身所能表达的范围区间所决定的。

读写锁能够实现锁降级，它按照先获取写锁、获取读锁再释放写锁的次序执行锁操作，而且写锁能够降级为读锁。

2.ReentrantReadWriteLock的类结构

从UML图可以看出ReentrantReadWriteLock中有多个内部类，这些内部类与ReentrantLock有些类似，都有Sync、NofairSync、FairSync,前者Sync继承于AQS,前者Sync是NofairSync、FairSync的父类。

与 ReentrantLock相比，虽然这3个静态内部类名字相同，但内部却有差异。ReentrantLock是排他锁，又可以说是一种写锁，这三个静态内部类只重写了AQS中的tryAcquire(int) 、tryRelease(int)这两个排他锁相关方法。而ReentrantReadWriteLock是读写锁，要实现排他锁和共享锁这两种锁，Sync、NofairSync、Fair不仅重写SynctryAcquire(int) 、tryRelease(int)方法，另外还重写了AQS中的tryAcquireShared(int)、 tryReleaseShared(int)这两个共享锁相关方法。

ReadLock是表示读锁的静态内部类，它主要委托NofairSync/FairSync中的共享锁相关方法实现的，如"void acquire(int)" "boolean release(int)"等（这两个方法是父类AQS的模板方法，模板方法再去调用自身重写的tryAcquire(int) 、tryRelease(int)方法）。

WriteLock是表示写锁的静态内部类，它主要委托NofairSync/FairSync中的排他锁相关方法实现的，如"void acquireShared(int)" "boolean releaseShared(int）"等。

HoldCounter和ThreadLocalHoldCounter又都是Sync的静态内部类，HoldCounter类的主要作用是记录获一个获取取到共享锁的读线程的重入次数，ThreadLocalHoldCounter继承于线程局部变量ThreadLocal,主要是为每个获取到共享锁的读线程单独维护一个HoldCounter。

 

Sync中几个值得注意的成员变量

    private transient ThreadLocalHoldCounter readHolds;

    private transient HoldCounter cachedHoldCounter;

    private transient Thread firstReader = null;
    
    private transient int firstReaderHoldCount;

 readHolds表示当前线程持有的可重入读锁的数量。 仅在Sync的构造方法和readObject方法中初始化，每当线程的读锁重入计数减少至0时将其移除。

cachedHoldCounter表示最后一个成功获取读锁的线程的重入次数计数器。 在下一个要释放的线程是最后一个要获取的线程的常见情况下，这可以节省在ThreadLocalHoldCounter中查找的时间。

firstReader表示第一个获得读取锁定的线程。firstReader是唯一一个最后一次将读锁重入计数从0更改为1且此后没有释放读取锁的线程； 如果没有这样的线程，则firstReader为null。firstReader使得读取追踪适用于非竞争读锁的效率更高。

firstReaderHoldCount表示第一个成功获取读锁的线程的重入次数。

3.ReadWriteLock的一些方法与使用示例

ReadWriteLock接口只有两个抽象方法，分别获取读锁和读锁。

public interface ReadWriteLock {
    /**
     * Returns the lock used for reading.
     *
     * @return the lock used for reading
     */
    Lock readLock();

    /**
     * Returns the lock used for writing.
     *
     * @return the lock used for writing
     */
    Lock writeLock();
}

另外ReentrantReadWriteLock类还有一些使我们更容易并发编程的一些辅助方法，这些方法都直接委托Sync（或其子类NofairSync/FairSync）去实现。

    public int getReadLockCount() {
        return sync.getReadLockCount();
    }

    public boolean isWriteLocked() {
        return sync.isWriteLocked();
    }
  
    public boolean isWriteLockedByCurrentThread() {
        return sync.isHeldExclusively();
    }
 
    public int getWriteHoldCount() {
        return sync.getWriteHoldCount();
    }
 
    public int getReadHoldCount() {
        return sync.getReadHoldCount();
    }

 getReadLockCount（）返回成功获取读锁的线程数，即此锁持有的读取锁的数量。。

isWriteLocked（）返回写锁是否被某个线程成功获取了。

isWriteLockedByCurrentThread（）返回写锁是否被当前线程成功获取了。

getWriteHoldCount（）返回当前线程重复获取写锁（重入）的次数。

getReadHoldCount（）返加当前线程重复获取读锁（重入）的次数。

使用示例

下面的代码使用HashMap去缓存简称/全称对应关系，addName()方法是写操作，因此使用写锁，getFullName()方法是读操作，因而使用读锁。

package juc;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class WRDemo {
    private final ReadWriteLock wrLock = new ReentrantReadWriteLock(false);
    private final Lock readLock = wrLock.readLock();
    private final Lock writeLock = wrLock.writeLock();
    private final Map<String, String> cacheForShortName = new HashMap<>();

    public boolean addName(String shortName, String fullName) {
        final Lock wl = writeLock;
        wl.lock();
        try {
            if (!cacheForShortName.containsKey(shortName)) {
                cacheForShortName.put(shortName, fullName);
                return true;
            }
            return false;
        } finally {
            wl.unlock();
        }
    }

    public String getFullName(String shortName) {
        final Lock rl = readLock;
        rl.lock();
        try {
            return cacheForShortName.get(shortName);
        } finally {
            rl.unlock();
        }
    }

    public static void main(String[] args) {
        final WRDemo wrDemo = new WRDemo();
        new Thread(() -> {
            boolean flag = wrDemo.addName("bj", "Beijing");
            System.out.println("添加 bj-Beijing:" + (flag ? "成功" : "失败"));

        }).start();
        new Thread(() -> {
            boolean flag = wrDemo.addName("sh", "Shanghai");
            System.out.println("添加 sh-Shanghai:" + (flag ? "成功" : "失败"));

        }).start();
        new Thread(() -> {
            System.out.println( "上海"+wrDemo.getFullName("sh"));

        }).start();
        new Thread(() -> {
            System.out.println("上海"+ wrDemo.getFullName("sh"));

        }).start();
        new Thread(() -> {
            boolean flag = wrDemo.addName("bj", "Beijing");
            System.out.println("第二次添加 bj-Beijing:" + (flag ? "成功" : "失败"));

        }).start();
    }
}

4.取/存读写状态的设计理念

ReentrantLock使用AQS中的int类型的state成员变量来保存排他锁状态，重入一次使state自增1，每释放一次重入状态state自减1（以前的帖子）。ReentrantReadWriteLock的重入设计理念应该也是基于此。ReentrantLock只需要保存写锁状态，直接使用state成员变量就可以实现，关键在于如何同时保存ReentrantReadWriteLock的读锁状态与写锁状态。其实可以借鉴CPU的标志寄存器的设计理念，将一个变量"按位分割"，不同的位范围表示不同的含义。ReentrantReadWriteLock就是将state的高16位表示读状态、低16位表示写状态（int类型4个字节，共32比特位）。

读取读写状态

利用位运算操作，可以分别将高16位和低16位的值取出来。

取低16位的写状态，只需要将state的高16位的所有位设为0即可，根据“0和（1或0）进行按位与操作结果都是0、（0或1）和1进行与操作结果均为其本身”的特点，可以使用按位与操作表达式"state&0x0000FFFF"实现，所以在Sync中exculusiveCount（）方法体中有"c&EXCLUSIVE_MASK",而EXCLUSIVE_MASK等于0x0000FFFF。

取高16位的读状态，只需要将state的低16位的抹除即可，可将state无符号右移16位，所以Sync中的sharedCount方法体中有代码"c >>> SHARED_SHIFT"。

        static final int SHARED_SHIFT   = 16;
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

        /** Returns the number of shared holds represented in count  */
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
        /** Returns the number of exclusive holds represented in count  */
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

写入读写状态

假设当前同步状态值为S， 由于低16位表示写锁状态，当写锁状态增加1时，只需要将state设为S+1即可，而高16位表示读锁状态，需要在state的第17位加1，即将state设为S+0x00010000.

因些在尝试获取写锁tryAcquire（）方法中有代码“compareAndSetState(c, c + acquires)”，在尝试获取读锁tryAcquireShared方法中有代码“compareAndSetState(c, c + SHARED_UNIT))”

   protected final int tryAcquireShared(int unused) {
       //....
       if (!readerShouldBlock() &&
               r < MAX_COUNT &&
               compareAndSetState(c, c + SHARED_UNIT)) {//SHARED_UNIT=0x00010000

       }
       //....
   }
    protected final boolean tryAcquire(int acquires) {
       //....省略
        if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))//acquires=1
            return false;
        //....省略
    }

 5.写锁的获取与释放

 写锁的lock()方法实际调用Sync的父类AQS的acquire（int)方法，acquire（int)是模板方法，acquire（int)的主要逻辑在以前的帖子中分析过，我们重点关注被Sync重写的tryAcuqrie(int)方法。

    public void lock() {
        sync.acquire(1);
    }

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

    尝试获取写锁

tryAcquire的基本逻辑：如果读锁被获取了或写锁被其他线程获取了，那么尝试获取写锁失败。如果在之前当前线程已经获取了写锁，增加重入次数，尝试获取写锁成功。若未有任何读锁、写锁被获取，则进行CAS更新state，
若更新成功,尝试获取写锁成功，返回true,若更新失败，尝试获取写锁失败，返回false.

    protected final boolean tryAcquire(int acquires) {
        /*
         * Walkthrough:
         * 1. If read count nonzero or write count nonzero
         *    and owner is a different thread, fail.
         * 2. If count would saturate, fail. (This can only
         *    happen if count is already nonzero.)
         * 3. Otherwise, this thread is eligible for lock if
         *    it is either a reentrant acquire or
         *    queue policy allows it. If so, update state
         *    and set owner.
         */
        Thread current = Thread.currentThread();
        int c = getState();
        int w = exclusiveCount(c);//写锁重入的次数
        if (c != 0) {//当前锁至少持有1个读锁或1个写锁（任何情况下最多只能有一个写锁）
            // (Note: if c != 0 and w == 0 then shared count != 0)
            /**
             *  因为 共享锁重入次数+排他锁重入次数 < state ,即 shareCout+ w<c,而又c>0 && w=0,
             *  那么shareCount>0,当前读锁被某线程获取了,所以这里w=0表示当前读锁被某些线程获取了，
             *  在读锁被获取了的情况下，不能获取写锁（只有在所有读锁、写锁均补充释放才能获取写锁），返回false。
             *
             *  "current != getExclusiveOwnerThread()"为true表明，
             *   前置条件"w==0"不成立，那么w>0，即写锁已经被某线程获取到了，
             *   而"current != getExclusiveOwnerThread()"条件本身又表明，当前线程不是获取到写锁的线程
             *   所以尝试获取写锁失败,返回false
             *
             *   综合起来说，当有读锁被某些线程成功获取或写锁被其他线程成功获取时，
             *   尝试获取写锁失败,返回false。
             */
            if (w == 0 ||  current != getExclusiveOwnerThread())
                return false;
            if (w + exclusiveCount(acquires) > MAX_COUNT) //超出了最大可重入次数,MAX_COUNT=0x0000FFFF
                throw new Error("Maximum lock count exceeded");
            // Reentrant acquire
             //之前写锁已经被当前线程成功获取,重入次数自增
            //尝试获取写锁成功,返回true
            setState(c + acquires);
            return true;
        }
        
        //getState=0，当前锁不持有任何读锁、写锁
        
        if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
            return false; //cas更新失败，尝试获取写锁失败，返回false
        //cas更新成功，设置当前线程为独占线程，尝试获取写锁成功，，返回false
        setExclusiveOwnerThread(current);
        return true;
    }

 这里与ReentrantLock不同的地方在于多了对读锁是否存在的判断。读锁的读取线程进行读取操作时，它不能主动感知写入操作，如果同时进行读写操作，读入的数据可能是被删除的数据或是过期的数据，读取时不能写入，所以在已有读锁的时候不能再去获取写锁。反过来也是一样，为保证数据的一致性，在有写锁的时候，它要阻塞其他所有的读写操作，不能再去获取任何读锁和写锁。

    尝试释放写锁

tryRelease(int)尝试释放锁的基本逻辑和ReentrantLock几乎一样：将重入次数自减，当重入次数为0，将独占线程设为null,返回true,反之返回false.

        protected final boolean tryRelease(int releases) {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            int nextc = getState() - releases;
            boolean free = exclusiveCount(nextc) == 0;
            if (free)
                setExclusiveOwnerThread(null);
            setState(nextc);
            return free;
        }

 6.读锁的获取与释放

读锁是一个支持重进入的共享锁，它能够被多个线程同时获取，在没有其他写线程访问（或者写状态为0）时，读锁总会被成功地获取，而所做的也只是（线程安全的）增加读状态。如果当前线程已经获取了读锁，则增加读状态。如果当前线程在获取读锁时，写锁已被其他线程获取，则进入等待状态。

   尝试获取读锁

tryAcquireShared(int)的主要逻辑：1.如果另一个线程持有写锁定，则失败。 2.否则，此线程符合请求获取读锁的条件，因而进一步询问于根据队列策略是否应该阻塞请求读锁的线程。 如果不是，请尝试按CAS更新state和更新相关重入次数的计数器。此步骤不检查重入获取，这将推迟到完整版本fullTryAcquireShared（Thread)方法，以避免在更典型的非重入情况下检查重入次数的计数。 3.如果第2步失败，表明第2步中应该阻塞读线程或重入次数饱和或CAS失败，此时进入进入fullTryAcquireShared（）方法进行完整版的读锁获取重试。

    protected final int tryAcquireShared(int unused) {
        /*
         * Walkthrough:
         * 1. If write lock held by another thread, fail.
         * 2. Otherwise, this thread is eligible for
         *    lock wrt state, so ask if it should block
         *    because of queue policy. If not, try
         *    to grant by CASing state and updating count.
         *    Note that step does not check for reentrant
         *    acquires, which is postponed to full version
         *    to avoid having to check hold count in
         *    the more typical non-reentrant case.
         * 3. If step 2 fails either because thread
         *    apparently not eligible or CAS fails or count
         *    saturated, chain to version with full retry loop.
         */
        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)) {//不应该阻塞读线程且CAS更新state成功
            if (r == 0) {  //之前读锁没有被任何线程获取
                firstReader = current; //将第一个获取到读锁的线程设为当前线程
                firstReaderHoldCount = 1; //第一个获取到读锁的线程重入次数初始为1.
            } else if (firstReader == current) { //当前线程之前已经获取到读锁，且是第一个获取到读锁的线程
                firstReaderHoldCount++;//重入次数自增1
            } else {
                //读锁被某些线程获取了，且当前线程不是第一个获取到读锁的线程
                HoldCounter rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))//当前线程不是最后获取到读锁的线程
                    //当线程本地变量readHolds中的属于当前线程的HoldCounter赋给cachedHoldCounter
                    cachedHoldCounter = rh = readHolds.get();
                else if (rh.count == 0)//当前线程是最后获取到读锁的线程且重入的次数是0
                    readHolds.set(rh);//将cachedHoldCounter设置为readHolds的当前线程变量
                rh.count++;//重入次数自增
            }
            return 1;
        }
        //当应该阻塞读线程或重入次数饱和或CAS失败，进入完整版的读锁获取重试
        return fullTryAcquireShared(current);
    }

fullTryAcquireShared()完整版的获取读，可处理tryAcquireShared中CAS失败和被忽略的可重入读

    final int fullTryAcquireShared(Thread current) {
        /*
         * This code is in part redundant with that in
         * tryAcquireShared but is simpler overall by not
         * complicating tryAcquireShared with interactions between
         * retries and lazily reading hold counts.
         */
        HoldCounter rh = null;
        for (;;) {
            int c = getState();
            if (exclusiveCount(c) != 0) {
                if (getExclusiveOwnerThread() != current)
                    return -1; ////写锁被其他线程成功获取了，尝试获取读锁失败。
                // else we hold the exclusive lock; blocking here
                // would cause deadlock.
            } else if (readerShouldBlock()) { //写锁未被任何线程获取，且不应该阻塞读线程
                // Make sure we're not acquiring read lock reentrantly
                if (firstReader == current) {//确保读锁不会申请重入
                    // assert firstReaderHoldCount > 0;
                } else {//当前线程不是第一个获取到读锁的线程
                    //更新cachedHoldCounter readHolds
                    if (rh == null) {
                        rh = cachedHoldCounter;
                        if (rh == null || rh.tid != getThreadId(current)) {
                            rh = readHolds.get();
                            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(int)的代码逻辑类似
            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;
            }
        }
    }

 fullTryAcquireShared与tryAcquireShared中的代码部分很相似，但由于不使tryAcquireShared与重试和延迟读取“hold counts”之间的交互复杂化，因此整体上更简单。

    尝试释放读锁

tryReleaseShared（int）,读锁的每次释放均减少读状态，减少的值是（1<<16）。与排他锁的最大不同在于，此处在更新state时，使用CAS更新，而不是无条件更新state,这是从线程安全的的角度考虑的，因为可能有多个读线程同时释放读锁。

    protected final boolean tryReleaseShared(int unused) {
        Thread current = Thread.currentThread();
        if (firstReader == current) {
            //如果当前线程是第一个读锁线程，将其重入次数自减，当重入次数为0时，firstReader赋空
            // assert firstReaderHoldCount > 0;
            if (firstReaderHoldCount == 1)
                firstReader = null;
            else
                firstReaderHoldCount--;
        } else {
            //更新cachedHoldCounter readHolds，重入次数自减，当重入次数为0时，
            // 将当前线程的HoldCounter从线程本地变量readHolds中移除
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                rh = readHolds.get();
            int count = rh.count;
            if (count <= 1) {
                readHolds.remove();
                if (count <= 0)
                    throw unmatchedUnlockException();
            }
            --rh.count;
        }
        for (;;) {
            /**
             * cas更新state,读锁重入次数减1，不能使用无条件更新state.
             * 这里是从线程安全的的角度考虑的，因为可能有多个读线程同时释放读锁。
             */
            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;
        }
    }