【源码解析】聊聊ReentrantReadWriteLock是如何实现的读写锁

为什么需要读写锁

在并发编程领域，有多线程进行提升整体性能，但是却引入了共享数据安全性问题。基本就是无锁编程下的单线程操作，有互斥同步锁操作，但是性能不高，并且同一时刻只有一个线程可以操作资源类。但是对于大多数常见下，都是读操作多，写操作少，那么可以利用将锁的粒度进行细化，进而分化出读锁/写锁。也就是syn/ReentrantLock的升级版本ReentrantReadWriteLock。

之前一篇文章已经简单介绍过 ，本篇主要从源码角度剖析具体原理如何实现的。
聊聊ReentrantReadWriteLock锁降级和StampedLock邮戳锁

源码解析

带着三个问题去梳理

• 读写锁是怎样实现分别记录读写锁的状态？
• 读锁如何获取和释放锁
• 写锁如何获取和释放锁

可以看到顶层通过接口定义规范，内部持有Sync实现AQS，分别实现不同的公平锁和非公平锁。

//读写锁的接口规范
public interface ReadWriteLock {

    Lock readLock();

    Lock writeLock();
}

// 内部持有读写锁 
public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    private static final long serialVersionUID = -6992448646407690164L;
   
    private final ReentrantReadWriteLock.ReadLock readerLock;
   
    private final ReentrantReadWriteLock.WriteLock writerLock;
    
    final Sync sync;

    public ReentrantReadWriteLock() {
        this(false);
    }

默认是非公平锁。内部通过构造方法创建两个锁，读锁和写锁。

    public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }

锁状态

看到这里其实有点懵逼，什么 这都是什么操作，其实在AQS内部通过一个变量state进行控制是否可以获取资源，但是读写锁如何要用两个变量的话，其实不太好，所以就通过高16位代表读锁的状态、低16位代表写锁的状态。

对于低16来说，值等于0没有加写锁，值等于1 加了写锁，大于1 标识写锁的重入次数。
高16来说，0 ：没有加读锁， 1: 加读锁。 值大于1 不表示读锁的重入次数，表示读锁总共被获取了多少次。读锁的重入次数存储在和线程相关的地方，通过threadLocal进行存储。

  abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 6317671515068378041L;

          // 偏移位数
        static final int SHARED_SHIFT   = 16;
        // 共享锁基本单位  左移16位 state+= shared_unit
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        // 读锁、写锁 可重入最大数量
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
        // 获取低16位的条件
        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  */
        // 写锁 是否持有 1 为一个线程持有 2 1次冲入 1次获取写锁
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

写状态，等于 S & 0x0000FFFF(将高 16 位全部抹去)。 当写状态加1，等于S+1.
读状态，等于 S >>> 16 (无符号补 0 右移 16 位)。当读状态加1，等于 S+(1<<16),也就是S+0x00010000。

这样 我们就完成了一个state值可以同时表示两种状态的。

写锁

写锁加锁

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

调用AQS的获取

    public final void acquire(int arg) {
        //tryAcquire(arg) true 获取锁成功直接结束
        //如果没有获取到锁，acquireQueued 会将线程压入队列中
        //!tryAcquire(arg)  没有获取到锁，将当前线程挂起
        //addWaiter
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

ReentrantReadWriteLock内部实现了tryAcquire方法。
该方法主要的作用就是
1.获取当前线程
2.判断state的状态。 c = 0 说明当前没有读锁和写锁，通过CAS进行设置state的值 直接获取锁
3.state值不等于0，w == 0 说明当前有读锁 获取锁失败，返回
4.w != 0 说明 当前是写锁重入，所以判断是否最大值，设置state的值+1
writerShouldBlock() 方法会根据是否是公平锁进行排队处理

       protected final boolean tryAcquire(int acquires) {
            // 获取当前线程
            Thread current = Thread.currentThread();
            // 获取state的值
            int c = getState();
            int w = exclusiveCount(c);
            // c = 0 说明 当前没有读锁和写锁
            if (c != 0) {
                // w == 0 等于0 说明 说明当前有读锁  或者当前线程不等于持有锁的线程
                // 写读互斥
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                // 获取写锁 不大于最大值
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                // Reentrant acquire
                // 设置当前值 说明可重入
                setState(c + acquires);
                return true;
            }
            // 是否需要阻塞 公平锁
            if (writerShouldBlock() ||
                    //CAS 设置c的值 c += 1
                !compareAndSetState(c, c + acquires))
                return false;
            // 设置为当前线程
            setExclusiveOwnerThread(current);
            return true;
        }

写锁释放锁

当当前线程执行完毕业务逻辑之后，就会释放锁。

        public void unlock() {
            sync.release(1);
        }

    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                //唤醒阻塞等待的线程
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

释放锁的流程主要就是
1.判断持有锁的线程是否属于当前线程，不是直接异常
2.将state-1 ,state = 0的话，说明重入的锁释放完毕。清空
3.设置state的值，可能是-1 或者 为0。

        protected final boolean tryRelease(int releases) {
            // 持有锁的线程 是否等于当前线程
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            // 将当前state -= 1
            int nextc = getState() - releases;
            boolean free = exclusiveCount(nextc) == 0;
            // 如果写锁为0 说明当前没有锁持有了
            if (free)
                // 将当前线程释放
                setExclusiveOwnerThread(null);
            // 设置state的值
            setState(nextc);
            return free;
        }

读锁

读锁加锁

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

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

        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 &&
                    //CAS 设置  高16位加1
                compareAndSetState(c, c + SHARED_UNIT)) {
                // 第一次获取读锁
                if (r == 0) {
                    //设置第一个获取读锁的线程
                    firstReader = current; // 当前线程
                    //设置第一个获取读锁线程的重入数
                    firstReaderHoldCount = 1; //
                } else if (firstReader == current) {
                    // 如果当前线程是第一个获取读锁的线程，重入数++
                    firstReaderHoldCount++;
                } else {
                    //刷新除获取锁的第一个读线程的重入数
                    // threadLocal进行记录线程重入次数
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;
            }
            // 再次尝试获取读锁，
            return fullTryAcquireShared(current);
        }

从这里可以看到，支持锁降级，持有写锁的线程，可以获取读锁，但是后续要记得把读锁和写锁读释放

读锁释放锁

        public void unlock() {
            sync.releaseShared(1);
        }

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

        protected final boolean tryReleaseShared(int unused) {
            Thread current = Thread.currentThread();
            // 如果当前线程是第一个获取读锁的线程
            if (firstReader == current) {
                // 第一个获取读锁的线程 重入次数等于=1
                // assert firstReaderHoldCount > 0;
                if (firstReaderHoldCount == 1)
                    //第一个获取读锁的线程设置为null
                    firstReader = null;
                else
                    // 当前线程重入多次 -1
                    firstReaderHoldCount--;
                //如果不是第一个获取读锁的线程，获取该线程的锁重入次数对象
            } else {
                // 获取线程持有共享锁的数量对象
                HoldCounter rh = cachedHoldCounter;
                // 如果rh==null 当前线程不是共享锁数量对象对应的线程id
                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;
            }
            //CAS同步更新
            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;
            }
        }

线程读锁的重入数与读锁数量是两个概念，线程读锁的重入数是每个线程获取同一个读锁的次数，读锁数量则是所有线程的读锁重入数总和。举一个例子就是 3个线程 分别获取了3次读锁，那么读锁数量就是9，每个线程的读锁重入数就是3。

锁升级&锁降级

锁升级就是线程持有读锁的前提下，去升级为写锁，显然这是违背读写互斥的。

锁降级，线程持有写锁的前提下，降级为读锁。

好了我们来看为什么需要锁降级，如果说针对一块临界区直接加一把大锁，那么其实并发读很低，那么可不可以在获取写锁的前提下 降级为读锁，这样既保证数据的一致性，又可以提升整体的并发度。锁降级就是为了结局这个问题。

设计思想

通过本篇的大概学习，我们了解到RRW中几个设计要点，通过一个变量去控制两个读写锁的状态，位运算的方式。值得我们借鉴，另一种就是锁降级的为了保证数据安全。以及在整体的代码实现上大量使用模板模式，AQS的子类都是相同的方式。