ReentrantReadWriteLock源码解析

末蓝、 2022-05-07 20:44 269阅读 0赞

`ReentrantReadWriteLock` 使用一个 `32` 位的 `int` 类型来表示锁被占用的线程数，高 `16` 位用来表示读锁占有的线程数量，低 `16` 位表示写锁被同一个线程申请次数。

// 读锁占用位数高16位 static final int SHARED_SHIFT = 16; // 增加读锁数量 static final int SHARED_UNIT = (1 << SHARED_SHIFT); // 申请读锁最大线程数量（65535） static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1; // 写锁重入次数 static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; // 当前线程持有读锁的重入次数，只在构造和readObject中初始化重入次数为0时删除 private transient ThreadLocalHoldCounter readHolds; // 最后一个获取读锁线程的重入计数 private transient HoldCounter cachedHoldCounter; // 第一个获得读锁的线程 private transient Thread firstReader = null; // 第一个获得读锁的线程重入计数 private transient int firstReaderHoldCount;

`ReentrantReadWriteLock` 有公平锁与非公平锁实现，这里主要看非公平锁实现。

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

### 读锁加锁 ###

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

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

// ReentrantReadWriteLock protected final int tryAcquireShared(int unused) { Thread current = Thread.currentThread(); int c = getState(); // 独占锁的重入次数不等于0 且 独占锁持有线程不等于当前线程，则返回-1获取共享锁失败 if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1; int r = sharedCount(c); // 持有共享锁的线程数量 // 判断获取读锁的线程是否需要阻塞 且 持有共享锁线程数量小于最大线程数量 且 CAS设置持有共享锁线程数量成功 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; // 最后一个获取读锁线程的重入计数为空 或 其线程id与当前线程id不一致，则将当前线程的重入次数设置为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); // CAS失败或其它原因则调用此方法处理 }

// ReentrantReadWriteLock final int fullTryAcquireShared(Thread current) { HoldCounter rh = null; for (;;) { int c = getState(); // 独占锁的重入次数不等于0 且 独占锁持有线程不等于当前线程，则返回-1获取共享锁失败 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; // 最后一个获取读锁线程的重入计数为空 或 其线程id与当前线程id不一致 if (rh == null || rh.tid != getThreadId(current)) { rh = readHolds.get(); // 当前线程持有读锁的重入次数等于0，则移除改计数对象 if (rh.count == 0) readHolds.remove(); } } // 重入次数为0，返回-1表示获取共享锁失败 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)) { 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; } } }

### 读锁释放 ###

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

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

// ReentrantReadWriteLock protected final boolean tryReleaseShared(int unused) { Thread current = Thread.currentThread(); // 判断第一个获取读锁的线程与当前线程是否相同 if (firstReader == current) { // assert firstReaderHoldCount > 0; // 如果重入次数为1，则直接置空firstReader if (firstReaderHoldCount == 1) firstReader = null; else firstReaderHoldCount--; // 重入次数自减 } else { HoldCounter rh = cachedHoldCounter; // 最后一个获取读锁线程的重入计数为空 或 其线程id与当前线程id不一致 if (rh == null || rh.tid != getThreadId(current)) rh = readHolds.get(); int count = rh.count; // 如果当前线程的锁重入次数<=1，则直接移除该计数对象 if (count <= 1) { readHolds.remove(); if (count <= 0) throw unmatchedUnlockException(); } --rh.count; // 自减锁重入次数 } for (;;) { int c = getState(); int nextc = c - SHARED_UNIT; // CAS设置同步state if (compareAndSetState(c, nextc)) return nextc == 0; } }

### 写锁加锁 ###

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

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

// ReentrantReadWriteLock.Sync protected final boolean tryAcquire(int acquires) { Thread current = Thread.currentThread(); int c = getState(); // 获取当前同步状态 int w = exclusiveCount(c); // 获取独占锁的重入次数 // c不等于0说明当前读锁或写锁被占用 if (c != 0) { // (Note: if c != 0 and w == 0 then shared count != 0) // 独占锁重入次数等于0 或 独占锁持有线程与当前线程不相同，则返回false表示获取写锁失败 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; } // 判断写锁是否需要阻塞 或者 CAS设置state失败，返回false获取写锁失败 if (writerShouldBlock() || !compareAndSetState(c, c + acquires)) return false; setExclusiveOwnerThread(current); // 设置独占锁持有对象为当前线程 return true; // 获取锁成功 }

### 写锁释放 ###

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

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

// ReentrantReadWriteLock.Sync protected final boolean tryRelease(int releases) { // 判断独占锁持有线程与当前线程是否一致，不一致则抛出异常 if (!isHeldExclusively()) throw new IllegalMonitorStateException(); int nextc = getState() - releases; // 递减写锁重入次数 boolean free = exclusiveCount(nextc) == 0; // 根据锁重入次数是否等于0判断锁释放标识 // 已经释放，则置空独占锁持有线程 if (free) setExclusiveOwnerThread(null); setState(nextc); // 设置释放后的锁重入次数 return free; }