@[toc]

概述

在上一篇博文jdk11源码-ReentrantLock源码中介绍了ReentrantLock的源码实现。ReentrantReadWriteLock是ReentrantLock的兄弟类，顾名思义，读写锁。当前其内部实现也是借助于AQS队列，不过与ReentrantLock的实现稍有不同，后面会逐步分析。

一个典型使用案例：

class RWDictionary {
  private final Map<String, Data> m = new TreeMap<>();
  private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
  private final Lock r = rwl.readLock();
  private final Lock w = rwl.writeLock();

  public Data get(String key) {
    r.lock();
    try { return m.get(key); }
    finally { r.unlock(); }
  }
  public List<String> allKeys() {
    r.lock();
    try { return new ArrayList<>(m.keySet()); }
    finally { r.unlock(); }
  }
  public Data put(String key, Data value) {
    w.lock();
    try { return m.put(key, value); }
    finally { w.unlock(); }
  }
  public void clear() {
    w.lock();
    try { m.clear(); }
    finally { w.unlock(); }
  }
}}

读写标记的存储

在之前分析ReentrantLock时，讲述了线程获得锁的标记是在state上的，state=0表示没有被加锁，state=1表示加锁成功，state>1表示锁重入。
但是ReentrantReadWriteLock是读写锁，既要保存是否加锁，还要保存锁的类型，以及重入，只有state一个字段是无法满足的。所以ReentrantReadWriteLock对其进行了改造，一方面引入了HoldCounter来保存重入数量；一方面将state分为高低位，state是int类型，32位，高16位保存共享锁（读锁），低16位保存独占锁（写锁）。接下来看一下详细实现。

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

        //state被拆分为高低位，高16位保存共享锁（读锁），低16位保存独占锁（写锁）
        //这几个字段是为了区分state的高低位的，具体使用会在下面讲解
        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;

        //返回共享锁的数量。c无符号右移16位也就是高16位的值。
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
        // 返回独占锁的数量。EXCLUSIVE_MASK转为2进制是16个1，c是32位，进行与计算，高16位结果全为0，低16位结果与c的低16位相同，也就是c的低16位的值。
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

        //HoldCounter类主要用于读锁的可重入，记录了重入的次数:count。
        //HoldCounter会被包装在ThreadLocal中，是线程安全的，并且缓存在cachedHoldCounter变量中
        static final class HoldCounter {
            int count;          // 初始值为0
            // 这里使用id，而不是引用，目的是为了避免垃圾回收
            final long tid = LockSupport.getThreadId(Thread.currentThread());
        }

        //这里重写了initialValue方法，可以保证获取到的HoldCounter对象是同一个，不会重复创建。
        static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> {
            public HoldCounter initialValue() {
                return new HoldCounter();
            }
        }

        /**
         * 当前线程持有的 可重入读锁 的数量。
         * 仅在构造函数和readObject方法中初始化。
         * 当count的值降至0时删除。
        */
        private transient ThreadLocalHoldCounter readHolds;
        
        //缓存最后一个成功获取读锁的线程的HoldCounter 
        //通常来说，下一个要释放锁的线程就是最后一个获取锁的线程。这里使用cachedHoldCounter来单独保存，可以减少在ThreadLocal中查找的开销
        private transient HoldCounter cachedHoldCounter;

        //第一个获取读锁的线程，确切来说firstReader是最后一次将共享计数从0更改为1的唯一线程，并且从那时起就没有释放读锁定; 如果没有这样的线程，则返回null。
        private transient Thread firstReader;
        //第一个获取读锁线程的 hold count
        private transient int firstReaderHoldCount;

        Sync() {
            //初始化一个readHolds，其count默认是0
            readHolds = new ThreadLocalHoldCounter();
            setState(getState()); // 这一步目的是为了确保state的可见性
        }

//省略其他代码
}

上面介绍了关键的一些属性变量及业务逻辑。关键点有

state分为高16位和低16位，高16位保存共享锁（读锁），低16位保存独占锁（写锁）
使用HoldCounter 来记录读锁的重入次数，HoldCounter会被包装在ThreadLocal中，是线程安全的

另外，锁的获取与释放都是走的同一段代码逻辑。区别点在于AQS队列非空时的插入规则。

写锁的加解锁过程分析

writeLock.lock

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

调用AQS类的java.util.concurrent.locks.AbstractQueuedSynchronizer#acquire方法：

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

其中addWaiter和acquireQueued方法与ReentrantLock中分析的一样，这里就不在讲述。着重分析一下tryAcquire方。这里具体实现在java.util.concurrent.locks.ReentrantReadWriteLock.Sync#tryAcquire中：

@ReservedStackAccess
protectedfinal boolean tryAcquire(int acquires) {
    Thread current = Thread.currentThread();
    int c = getState();//获取state的值
    int w = exclusiveCount(c);//获取独占锁的数量
    if (c != 0) {
        // (Note: if c != 0 and w == 0 then shared count != 0)
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;//走到这里，c不等于0，w（低16位）是0，所以高16位肯定有值。也就是说此时有共享锁存在。
        if (w + exclusiveCount(acquires) > MAX_COUNT)//校验独占锁的数量是否超标
            throw new Error("Maximum lock count exceeded");

        //走到这里，说明当前是写锁重入。因为走到这一步只可能是：c不是0，w不是0，exclusiveOwnerThread标识的线程是当前线程。
        setState(c + acquires);
        return true;
    }
    
    //走到这里说明c的值是0。也就是读锁和写锁都没有。当然可以加锁了。
    if (writerShouldBlock() ||
        !compareAndSetState(c, c + acquires))
        return false;
    setExclusiveOwnerThread(current);//加锁成功，设置获取独占锁的线程为当前线程
    return true;
}

上面就是获取锁的过程，总结一下就是：

state的值是0，说明当前没有锁，那么判断写锁是否应该阻塞，如果不阻塞，则修改state的值，修改成功则加锁成功
state的值不是0，低16位是0，说明高16位不是0，此时有共享锁存在，不可以加锁。
state的值不是0，低16位不是0，但是当前获取独占锁的线程不是当前线程，不可以加锁
state的值不是0，低16位不是0，当前获取独占锁的线程是当前线程，说明是写锁重入，可以加锁

writeLock.unlock

writeLock.unlock源码：

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

他调用AQS类的release方法：

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

分析tryRelease方法，他的实现在ReentrantReadWriteLock.Sync类中，其他的与ReentranteLock中分析的一致

protected final boolean tryRelease(int releases) {
    if (!isHeldExclusively())//独占锁的持有者不是当前线程，抛异常
        throw new IllegalMonitorStateException();
    int nextc = getState() - releases;//这里是独占锁，低16位，所以可以直接相减，下一步会校验是不是低16位减为0
    boolean free = exclusiveCount(nextc) == 0;//判断低16位的独占锁是否都释放完
    if (free)
        setExclusiveOwnerThread(null);//独占锁已经释放完，exclusiveOwnerThread设为null
    setState(nextc);//释放state
    return free;
}

独占锁的释放过程比较简单的，首先检查加锁的是不是当前线程，不是则抛异常。接下来校验释放releases数量的state后，低16位是否为0，是，则说明独占锁释放完成，固设置exclusiveOwnerThread=null，否则设置新的state。

接下来看一下unparkSuccessor方法，写锁释放后，唤醒队列中后继节点，该后继节点可能是读锁也可能是写锁。

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        node.compareAndSetWaitStatus(ws, 0);//如果<0（这里一般是Node.PROPAGATE这种状态），重新将其waitstatus设置为0

    Node s = node.next;
    if (s == null || s.waitStatus > 0) {//如果后继节点被取消，那么找到后继第一个waitStatus <= 0的节点，将其唤醒
        s = null;
        for (Node p = tail; p != node && p != null; p = p.prev)
            if (p.waitStatus <= 0)
                s = p;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

公平锁与非公平锁

在上面tryAcquire代码中有一个方法writerShouldBlock来判断写锁是否应该被阻塞。这里公平锁和非公平锁的实现是不同的。

公平锁的实现：

static final class FairSync extends Sync {
        private static final long serialVersionUID = -2274990926593161451L;
        final boolean writerShouldBlock() {
            return hasQueuedPredecessors();
        }
        final boolean readerShouldBlock() {
            return hasQueuedPredecessors();
        }
    }
public final boolean hasQueuedPredecessors() {
        Node h, s;
        if ((h = head) != null) {
            if ((s = h.next) == null || s.waitStatus > 0) {
                s = null; // traverse in case of concurrent cancellation
                for (Node p = tail; p != h && p != null; p = p.prev) {
                    if (p.waitStatus <= 0)
                        s = p;
                }
            }
            if (s != null && s.thread != Thread.currentThread())
                return true;
        }
        return false;
    }

读锁的加解锁过程分析

readLock.lock

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

该方法会调用AQS的acquireShared方法

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

首先尝试获取tryAcquireShared，如果失败则执行doAcquireShared。

tryAcquireShared

tryAcquireShared的实现在ReentrantReadWriteLock中：

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)) {//不阻塞，并且申请一个读锁成功
        if (r == 0) {//共享锁数量为0：则记录最后一次将共享锁计数从0更改为1的线程，初始化firstReaderHoldCount =1
            firstReader = current;
            firstReaderHoldCount = 1;
        } else if (firstReader == current) {
            firstReaderHoldCount++;//firstReader 标识的第一个读线程 读锁重入
        } else {
            HoldCounter rh = cachedHoldCounter;//cachedHoldCounter：缓存的最后一个获取读锁的线程的HoldCounter 
            if (rh == null || rh.tid != LockSupport.getThreadId(current))//缓存为空，或者缓存的线程id不是当前线程
                cachedHoldCounter = rh = readHolds.get();//获取threadlocal中当前线程的计数器缓存HoldCounter ，并将其添加到缓存cachedHoldCounter中
            else if (rh.count == 0)//由于count为0时，readHolds中的缓存会被移除掉，所以这里要在set一下
                readHolds.set(rh);//将缓存的cachedHoldCountert添加到threadlocal变量readHolds中
            rh.count++;//当前线程重入计数器加一
        }
        return 1;
    }
    return fullTryAcquireShared(current);
}

可以看到这里入参没有用到，而是通过compareAndSetState(c, c + SHARED_UNIT)固定的申请一个读锁标记。整体流程如下：

在这里插入图片描述

==这里隐藏的优化点==：

firstReader 和firstReaderHoldCount。用于缓存第一个获取读锁的线程。目的是为了减少threadlocal的操作，提高效率。
cachedHoldCounter：用于缓存最后一个获取读锁的线程。当然这个也会同步保存到threadlocal变量readHolds中
将大概率执行成功的代码与通用代码分隔，提高性能（可能会降低一点可读性）。

当(!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)判断失败时，会进入fullTryAcquireShared()方法，该方法会采用CAS的方式来保证操作执行成功。
fullTryAcquireShared方法与上面tryAcquireShared逻辑基本相同，但是为什么也要写两份呢？其实这也是为了性能考虑。
读者在阅读java线程及锁相关的代码时，会经常遇到类似的逻辑，将通用方法与大概率出现的代码逻辑分开，对于大概率出现的逻辑进行针对性的优化来提高性能。
读者可以考虑将fullTryAcquireShared上面的代码去除，代码运行并不会受影响。

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 (firstReader == current) {
                // assert firstReaderHoldCount > 0;
            } else {
                if (rh == null) {
                    rh = cachedHoldCounter;
                    if (rh == null ||
                        rh.tid != LockSupport.getThreadId(current)) {
                        rh = readHolds.get();
                        if (rh.count == 0)
                            readHolds.remove();//count为0，当前线程的所有读锁都释放了，将其删除
                    }
                }
                if (rh.count == 0)
                    return -1;
            }
        }
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        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 != LockSupport.getThreadId(current))
                    rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
                cachedHoldCounter = rh; //缓存最后一个成功获取读锁的线程
            }
            return 1;
        }
    }
}

doAcquireShared

先看一下tryAcquireShared方法有哪些情况会返回<0

有独占锁存在，并且该独占锁不是当前线程加的。
没有独占锁存在，并且读锁需要等待时，并且当前线程不是第一个获取读锁的线程，并且最后一个获取锁的线程全部已经释放锁时。

执行到doAcquireShared这个方法，该线程加入AQS队列了，队列中的读锁成功获取锁后，需要唤醒后继节点，具体后面会详细介绍。

doAcquireShared: 以共享无中断模式获取。无限循环获取读锁，直到成功或者发生异常

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);//以共享模式添加到AQS队列尾部。
    boolean interrupted = false;
    try {
        for (;;) {
            final Node p = node.predecessor();//获取上一个node
            if (p == head) {//上一个节点是head节点，说明轮到自己获取锁了，则尝试获取读锁tryAcquireShared
                //翻阅源码，在ReentrantReadWriteLock中，这个返回值 r 要么是1，要么是-1
                //当然其他情况，可能返回0
                int r = tryAcquireShared(arg);
                if (r >= 0) {//获取读锁成功。
                    setHeadAndPropagate(node, r);
                    p.next = null; // 老的头结点已经没有用了，其next指向null，目的为了GC释放内存
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node))//检测是否应该park等待
                interrupted |= parkAndCheckInterrupt();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    } finally {
        if (interrupted)
            selfInterrupt();//被中断过，重置中断标记
    }
}

//设置头结点并且广播 唤醒后继节点
private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head;
    setHead(node);
    
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        
        Node s = node.next;
        if (s == null || s.isShared())//下一个节点如果为空，或者是共享模式的节点
            doReleaseShared();
    }
}

共享模式的Node在新建时，会设置nextWaiter=SHARE：

Node(Node nextWaiter) {
    this.nextWaiter = nextWaiter;
    THREAD.set(this, Thread.currentThread());
}

这里需要说一下tryAcquireShared的返回值。结合ReentrantReadWriteLock、CountDownLatch、Semaphore等实现，tryAcquireShared的返回值可能是0,1，-1，大于1 四种情况，其中大于1可以和1归为一类。
==tryAcquireShared的返回值大于0时，表示后继节点需要被唤醒；0表示加锁成功但是不需要唤醒后继节点；-1表示失败==

再分析一下setHeadAndPropagate方法唤醒规则(propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0)：

propagate>0表示后继节点需要被唤醒。
或者不论旧的头结点还是新的头结点，只要他的waitStatus<0那么就执行唤醒

这个waitStatus的设置参考shouldParkAfterFailedAcquire方法，其源码在上一篇文章jdk11源码-ReentrantLock源码已经分析过，核心就是在park时首先将其前驱节点的waitStatus设置为SIGNAL（-1）。所以waitStatus<0时就执行唤醒。

注意一点区别：如果tryAcquireShared成功，此时并没有加入AQS队列中，所以不会有通知唤醒之类的操作。而到了doAcquireShared方法中，首先是要现将其加入到AQS队列中，所以在获取所以后，就需要唤醒后继节点了。

doReleaseShared

private void doReleaseShared() {
    for (;;) {
        Node h = head;//注意这里已经是新的头节点了
        if (h != null && h != tail) {//头结点存在并且不等于tail，也就是AQS队列不为空的情况
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {//是SIGNAL状态，则需要unpark后继节点
                if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))//将头结点的waitstatue设置为0，以后就不会再次唤醒后继节点了。这一步是为了解决并发问题，保证只unpark一次！！
                    continue;            // 设置失败则进行重试
                unparkSuccessor(h);//唤醒头节点的后继节点
            }
            else if (ws == 0 && !h.compareAndSetWaitStatus(0, Node.PROPAGATE))//这里尝试将头结点设置为PROPAGATE。保证传播性
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        node.compareAndSetWaitStatus(ws, 0);//如果<0（这里一般是Node.PROPAGATE这种状态），重新将其waitstatus设置为0

    Node s = node.next;
    if (s == null || s.waitStatus > 0) {//如果后继节点被取消，那么找到后继第一个waitStatus <= 0的节点，将其唤醒
        s = null;
        for (Node p = tail; p != node && p != null; p = p.prev)
            if (p.waitStatus <= 0)
                s = p;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

读锁的特性之一：AQS队列中读锁获取成功后，需要将其后面的读锁唤醒。当后继读锁的节点唤醒后，他会重新申请读锁，此时如果成功，那么他会再次唤醒他的后继节点，如此往复，AQS队列中的头结点后的所有读锁都会被唤醒，如果遇到等待的写锁，终止传递。

总结读锁加锁过程：

acquireShared()申请锁，如果成功，则进入临界区继续执行。如果失败，则加入AQS等待队列，挂起（park）等待被唤醒（unpark）。
AQS队列中的读锁被唤醒后，重新尝试获取锁，如果成功，则进入临界区执行，并且唤醒后继共享节点，并且会传递下去一次唤醒所有的共享节点，直到队尾或者遇到写锁。

readLock.unlock

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

//releaseShared调用AQS类中的releaseShared方法
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {//尝试释放锁，ReentrantReadWriteLock中，只有读锁和写锁全部释放才会返回true
        doReleaseShared();//如果释放锁成功后，没有读锁也没有写锁，那么唤醒AQS的后继节点
        return true;
    }
    return false;
}

releaseShared方法很简单，先释放锁，在唤醒后继节点。

ReentrantReadWriteLock中，tryReleaseShared方法只有读锁和写锁全部释放才会返回true。这里注意：tryReleaseShared是一个接口，不同的锁的实现机制不同，比如信号量（Semaphore）机制下，只要释放资源成功就会返回true，而不会关心state的余额。这个在分析Semaphore源码时具体分析。

我们看ReentrantReadWriteLock中tryReleaseShared的实现：

protected final boolean tryReleaseShared(int unused) {
    Thread current = Thread.currentThread();
    if (firstReader == current) {//当前线程是第一个获取读锁线程，释放 firstReader 及firstReaderHoldCount 
        if (firstReaderHoldCount == 1)
            firstReader = null;
        else
            firstReaderHoldCount--;
    } else {
    //当前线程不是第一个读线程，需要从threadlocal中获取HoldCounter ，释放其count
        HoldCounter rh = cachedHoldCounter;
        if (rh == null ||
            rh.tid != LockSupport.getThreadId(current))
            rh = readHolds.get();
        int count = rh.count;
        if (count <= 1) {
            readHolds.remove();//说明这已经是最后一次释放锁了，直接删除即可
            if (count <= 0)
                throw unmatchedUnlockException();//说明锁计数器出现异常
        }
        --rh.count;//释放count的值
    }
    for (;;) {
        int c = getState();
        int nextc = c - SHARED_UNIT;//读锁标记减去 1 （还记得高16位表示读锁数量吗？）
        if (compareAndSetState(c, nextc))
            //当没有读锁也没有写锁时，返回true
            return nextc == 0;
    }
}

tryReleaseShared方法释放读锁占用的资源，释放以后，如果state!=0，说明有读锁或者写锁。那么直接返回。如果state==0，说明此事没有读锁也没有写锁了，需要唤醒AQS队列中等待的节点。

doReleaseShared有两个入口：一个是读锁释放资源后，一个是setHeadAndPropagate方法中。doReleaseShared的源码上面已经分析过。这里说一下不同之处：前面是读锁加锁后setHeadAndPropagate内部调用的doReleaseShared，他前面有一个判断s == null || s.isShared()，也就是说只能后继节点是读锁才可以唤醒后继节点。这也很好理解，一个读锁抢占资源成功，那么AQS队列中后面等待的读锁都应该唤醒进入临界区，这才符合读锁（共享锁）的特性吗。
然而在读锁释放资源时调用doReleaseShared方法时，此时唤醒的有可能是读锁，也可能是写锁，因为我们并不知道AQS队列中第一个节点是读还是写，所以根据唤醒的节点不同走不同的分支。

jdk11源码--ReentrantReadWriteLock源码

概述

读写标记的存储

写锁的加解锁过程分析

writeLock.lock

writeLock.unlock

公平锁与非公平锁

读锁的加解锁过程分析

readLock.lock

tryAcquireShared

doAcquireShared

doReleaseShared

readLock.unlock

Java技术进阶

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

jdk11源码--ReentrantReadWriteLock源码

概述

读写标记的存储

写锁的加解锁过程分析

writeLock.lock

writeLock.unlock

公平锁与非公平锁

读锁的加解锁过程分析

readLock.lock

tryAcquireShared

doAcquireShared

doReleaseShared

readLock.unlock

Java技术进阶

热门文章

最新文章

相关课程

相关电子书

相关实验场景