ReentrantReadWriteLock之ReadLock加锁解锁过程源码解析

写的很乱，ReentrantReadWriteLock调来调去的。不懂或者有错误欢迎留言

加锁

public void lock() { sync.acquireShared(1); } public final void acquireShared(int arg) { if (tryAcquireShared(arg) < 0) //加锁失败后调用这个方法。 doAcquireShared(arg); }

tryAcquireShared方法

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); //判断自己是否需要阻塞主要是可能有人要来上写锁。在排队 // 如果不需要阻塞，且r小于最大值，去cas加读锁 // 这个地方其实加了读锁，写锁就加不上了。写锁判断了一次。 if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) { //如果r == 0，表示第一个加锁的 if (r == 0) { //这个就是set操作 //这个地方其实主要是缓存， //读锁搞了两个缓存，第一个加锁的，和最后一个 firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) { firstReaderHoldCount++; } else { //这个holdCounter就维护了一个数量，和线程id //这个holdCounter是最后一次加锁的，上一次。 HoldCounter rh = cachedHoldCounter; //如果为空，或者id不相同再去threadLocal里面去拿 if (rh == null || rh.tid != getThreadId(current)) //放到缓存里面去 cachedHoldCounter = rh = readHolds.get(); // 表示这个东西是新拿出来的， //（一种是解锁后一种是第一次） else if (rh.count == 0) readHolds.set(rh); rh.count++; } //加读锁成功 return 1; } //cas失败，后者需要排队进入这个方法 return fullTryAcquireShared(current); }

readerShouldBlock方法

//非公平锁 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(); } //公平锁 final boolean readerShouldBlock() { return hasQueuedPredecessors(); }

apparentlyFirstQueuedIsExclusive方法

final boolean apparentlyFirstQueuedIsExclusive() { Node h, s; //如果头节点是空（队列未初始化）返回false //如果头节点的的下一个节点为null返回false //如果头节点的下一个节点的节点属性是共享的返回false //如果头节点的下一个节点的thread为null（猜测是取消了）返回false return (h = head) != null && (s = h.next) != null && !s.isShared() && s.thread != null; }

fullTryAcquireShared方法

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) { //线程不相同，返回-1，表示加锁失败 //什么时候会出现这种情况呢？ //tryAcquireShared下面cas失败后会出现这种情况。 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 { //不是的话，如果rh为空，第一次循环必为null if (rh == null) { rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) { //拿到当前threadLocal里面的HoldCounter 对象，注意这里不是0.上面+了1 rh = readHolds.get(); //如果这个地方count为0后面直接return-1了。 if (rh.count == 0) readHolds.remove(); } } if (rh.count == 0) return -1; } } //如果等于最大值的时候，抛出异常，溢出了 if (sharedCount(c) == MAX_COUNT) throw new Error("Maximum lock count exceeded"); //cas去加读锁。 if (compareAndSetState(c, c + SHARED_UNIT)) { //如果是0，你去看那个方法，他是拿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 } //返回1，枷锁成功。加锁失败继续循环，只要没有上写锁，就去cas加读锁。 return 1; } } }

doAcquireShared方法

private void doAcquireShared(int arg) { //addWaiter方法之后再ReentrantLock再去写。 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { boolean interrupted = false; for (;;) { //其实大部分和ReentrantLock类似。 final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg); //加锁成功 if (r >= 0) { //将当前node设到aqs队首，里面还会去让下一个进行加读锁 setHeadAndPropagate(node, r); p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); }

setHeadAndPropagate方法

private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // Record old head for check below setHead(node); /* * Try to signal next queued node if: * Propagation was indicated by caller, * or was recorded (as h.waitStatus either before * or after setHead) by a previous operation * (note: this uses sign-check of waitStatus because * PROPAGATE status may transition to SIGNAL.) * and * The next node is waiting in shared mode, * or we don't know, because it appears null * * The conservatism in both of these checks may cause * unnecessary wake-ups, but only when there are multiple * racing acquires/releases, so most need signals now or soon * anyway. */ //如果propagate > 0，因为是1.或者头节点是空，队列未初始化 //或者头节点waitStatus小于0，小于0说明后面有人等。 if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { //拿到他的后一个节点。 Node s = node.next; //如果s是空，或者s是共享节点（上读锁的）。 if (s == null || s.isShared()) doReleaseShared(); } }

doReleaseShared方法

private void doReleaseShared() { /* * Ensure that a release propagates, even if there are other * in-progress acquires/releases. This proceeds in the usual * way of trying to unparkSuccessor of head if it needs * signal. But if it does not, status is set to PROPAGATE to * ensure that upon release, propagation continues. * Additionally, we must loop in case a new node is added * while we are doing this. Also, unlike other uses of * unparkSuccessor, we need to know if CAS to reset status * fails, if so rechecking. */ for (;;) { Node h = head; //aqs队列初始化，且有人排队。 if (h != null && h != tail) { int ws = h.waitStatus; //如果他是-1，-1表示后面还有一个节点再排队 if (ws == Node.SIGNAL) { //将状态置回去。 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases //这个在写锁中介绍过了。 unparkSuccessor(h); } //如果ws == 0，说明后面没有人排队。将他改为-3. else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; } }

解锁过程

public void unlock() { sync.releaseShared(1); } public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }

tryReleaseShared方法

protected final boolean tryReleaseShared(int unused) { //没什么多的好说的。就是重入情况和非重入情况相减 Thread current = Thread.currentThread(); if (firstReader == current) { // assert firstReaderHoldCount > 0; if (firstReaderHoldCount == 1) firstReader = null; else firstReaderHoldCount--; } else { 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; } //死循环解锁。将持有读锁的线程数量-1. 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; } }