JDK 源码分析：ReentrantReadWriteLock
2011 年 12 月 9 日
abstract static class Sync extends AbstractQueuedSynchronizer { // 使用 AQS 中的 state 变量（int 类型）来记录读写锁的占用情况 // 其中高 16 位记录读锁的持有次数；低 16 位记录写锁的重入次数 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;
// 共享锁（读锁）的持有次数 static int sharedCount(int c) { return c >>> SHARED_SHIFT; } // 互斥锁（写锁）的重入次数 static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
/** * 每个线程持有读锁的计数器。 * 以 ThreadLocal 形式保存，缓存在 cachedHoldCounter * 该类的主要作用是记录线程持有读锁的数量，可理解为 的形式 */ static final class HoldCounter { int count = 0; // Use id, not reference, to avoid garbage retention final long tid = getThreadId(Thread.currentThread()); } static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> { public HoldCounter initialValue() { return new HoldCounter(); } }
/** * 当前线程持有的可重入读锁的数量（数量为0时删除）。 */ private transient ThreadLocalHoldCounter readHolds;
private transient HoldCounter cachedHoldCounter;
/** * firstReader：第一个获取读锁的线程； * firstReaderHoldCount：firstReader 的持有计数。 */ private transient Thread firstReader = null; private transient int firstReaderHoldCount;
Sync() { readHolds = new ThreadLocalHoldCounter(); setState(getState()); // ensures visibility of readHolds }
/* * 对于公平锁和非公平锁，获取和释放锁使用的代码相同； * 但在队列非空时，它们是否或如何允许插入的方式不同。 */ /** * 当前线程在尝试（或有资格）获取读锁时，是否应该由于策略原因而阻塞。 */ abstract boolean readerShouldBlock();
/** * 当前线程在尝试（或有资格）获取写锁时，是否应该由于策略原因而阻塞。 */ abstract boolean writerShouldBlock();
// 释放写锁 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; }
// 获取写锁 protected final boolean tryAcquire(int acquires) { /* * 流程： * 1. 若其他线程持有读锁或写锁（计数不为零），返回 false； * 2. 若持有数量饱和（超出上限），返回 false； * 3. 该线程有资格获取锁，更新 state 并设置为 owner。 */ Thread current = Thread.currentThread(); int c = getState(); 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; if (w + exclusiveCount(acquires) > MAX_COUNT) throw new Error("Maximum lock count exceeded"); // Reentrant acquire setState(c + acquires); return true; } // 若获取写锁时应该阻塞，或者更新 state 失败，返回 false if (writerShouldBlock() || !compareAndSetState(c, c + acquires)) return false; setExclusiveOwnerThread(current); return true; }
// 释放读锁 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; } // 更新 state 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; } }
// 获取读锁 protected final int tryAcquireShared(int unused) { /* * 流程： * 1. 如果其他线程持有写锁，获取失败； * 2. 否则，该线程有资格获取，因此请询问它是否由于队列策略而阻塞； * 若不阻塞，尝试通过 CAS 更新状态计数。 * 注意：这一步没有检查可重入的获取，推迟到完整版本， * 以避免在明显不可重入的情况下检查持有计数。 * 3. 如果第二步失败，要么是因为线程明显不符合条件、CAS 失败或计数饱和， * 则进行完整重试版本。 */ Thread current = Thread.currentThread(); int c = getState(); // step1. 若写锁被其他线程占用，则获取失败 // exclusiveCount(c) != 0表示写锁被占用 if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1; // step2. 获取读锁数量 int r = sharedCount(c); if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) { // 读锁未被占用，设置该线程是第一个持有读锁的线程 if (r == 0) { firstReader = current; firstReaderHoldCount = 1; // 该线程已持有读锁，计数加1 } else if (firstReader == current) { firstReaderHoldCount++; // 其他线程已持有读锁 } else { // 取缓存 HoldCounter rh = cachedHoldCounter; // 若未初始化，或者拿到的不是当前线程的计数，则从 ThreadLocal 中获取 if (rh == null || rh.tid != getThreadId(current)) cachedHoldCounter = rh = readHolds.get(); else if (rh.count == 0) readHolds.set(rh); // 增加计数 rh.count++; } // 获取成功 return 1; } // step3. 若step2获取失败，则执行该步骤 return fullTryAcquireShared(current); }
/** * 获取读锁的完整版，处理 tryAcquireShared 中未处理的 CAS 丢失和可重入读取。 */ final int fullTryAcquireShared(Thread current) { 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 { 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"); 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; } } }
/** * 执行写锁的 tryLock 方法 * 与 tryAcquire 相比，该方法未调用 writerShouldBlock */ final boolean tryWriteLock() { Thread current = Thread.currentThread(); int c = getState(); if (c != 0) { int w = exclusiveCount(c); if (w == 0 || current != getExclusiveOwnerThread()) return false; if (w == MAX_COUNT) throw new Error("Maximum lock count exceeded"); } if (!compareAndSetState(c, c + 1)) return false; setExclusiveOwnerThread(current); return true; }
/** * 执行读锁的 tryLock 方法 * 与 tryAcquireShared 相比，该方法未调用 readerShouldBlock */ final boolean tryReadLock() { Thread current = Thread.currentThread(); for (;;) { int c = getState(); if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return false; int r = sharedCount(c); if (r == MAX_COUNT) throw new Error("Maximum lock count exceeded"); if (compareAndSetState(c, c + SHARED_UNIT)) { if (r == 0) { firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) { firstReaderHoldCount++; } else { 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 true; } } }}
About The Author
maynard
懒散，不想无聊