本文共 9840 字,大约阅读时间需要 32 分钟。
这篇文章主要从源码角度讲解ReentrantReadWriteLock的ReadLock的加锁和减锁过程。
ReentrantReadWriteLock的ReadLock加锁解锁过程依赖于AQS类,所以有些相同的逻辑可以看看ReentrantLock的逻辑。
ReentrantReadWriteLock的ReadLock的唤醒过程具备传播性:
假设按照顺序A->B->C->D占用读锁,唤醒会依次进行 A线程占用读锁被唤醒后,A线程的锁释放会唤醒B线程。 B线程占用读锁被唤醒后,B线程的锁释放会唤醒C线程。 C线程占用读锁被唤醒后,C线程的锁释放会唤醒D线程。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) { Thread current = Thread.currentThread(); int c = getState(); // 如果当前锁写状态不为0且占锁线程非当前线程,那么返回占锁失败。 // 也就是当前线程先占写锁后可以再占读锁的,反之不行。 if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1; // 判断高位的读状态标记 int r = sharedCount(c); //如果公平策略没有要求阻塞且重入数没有到达最大值,则直接尝试CAS更新state //如果读不应该阻塞并且读锁的个数小于最大值65535,并且可以成功更新状态值 if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) { // 如果当前线程满足对state的操作条件, // 就利用CAS设置state+SHARED_UNIT,实际上就是读状态+1。 // 但是需要注意,这个state是全局的,即所有线程获取读锁次数的总和, // 而为了方便计算本线程的读锁次数以及释放掉锁, // 需要在ThreadLocal中维护一个变量。这就是HoldCounter。 //如果当前读锁为0 if (r == 0) { // 第一个读线程就是当前线程 firstReader = current; firstReaderHoldCount = 1; } //如果当前线程重入了,记录firstReaderHoldCount else if (firstReader == current) { firstReaderHoldCount++; } //当前读线程和第一个读线程不同,记录每一个线程读的次数 else { // 每个线程自己维护cachedHoldCounter HoldCounter rh = cachedHoldCounter; // 计数器为空或者计数器的tid不为当前正在运行的线程的tid 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); }fullTryAcquireShared过程
final int fullTryAcquireShared(Thread current) { HoldCounter rh = null; // 无限循环 for (;;) { // 获取状态 int c = getState(); // 写线程数量不为0且当前线程不是写线程那么返回获锁失败 if (exclusiveCount(c) != 0) { if (getExclusiveOwnerThread() != current) return -1; } // 写线程数量为0并且读线程被阻塞 else if (readerShouldBlock()) { 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; } } }doAcquireShared过程
private void doAcquireShared(int arg) { final Node node = addWaiter(Node.SHARED); boolean failed = true; try { boolean interrupted = false; // 开始自旋重试 for (;;) { // 获取当前线程代表节点的前一个节点 final Node p = node.predecessor(); // 如果前一个节点是head节点(head节点不保存任何线程), // 表明当前节点是第一个等待唤醒节点 if (p == head) { // 尝试获取锁 int r = tryAcquireShared(arg); // 如果获锁成功 if (r >= 0) { // 说明当前线程获取读锁成功,那么设置当前线程Node为head // 同时扩散唤醒相关读线程,因为读线程之间相互不阻塞,可以一起唤醒继续工作 setHeadAndPropagate(node, r); p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; } } // 如果“当前线程”不是CLH队列的表头, // 则通过shouldParkAfterFailedAcquire()判断是否需要等待, // 需要的话,则通过parkAndCheckInterrupt()进行阻塞等待。 // 若阻塞等待过程中,线程被中断过,则设置interrupted为true。 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }setHeadAndPropagate过程
private void setHeadAndPropagate(Node node, int propagate) { // 设置当前节点为新的head节点 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(); } }
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) { // 得到调用unlock的线程 Thread current = Thread.currentThread(); //如果是第一个获得读锁的线程,进行解锁 if (firstReader == current) { // assert firstReaderHoldCount > 0; if (firstReaderHoldCount == 1) firstReader = null; else firstReaderHoldCount--; } // 否则,线程ThreadLocal的HoldCounter中计数-1 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; } // 死循环 for (;;) { int c = getState(); // 释放一把读锁 int nextc = c - SHARED_UNIT; // 如果CAS更新状态成功,返回读锁是否等于0;失败的话,则重试 if (compareAndSetState(c, nextc)) return nextc == 0; } }doReleaseShared过程
private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { // 当前节点为SIGNAL那么就需要通过unparkSuccessor(p)唤醒后置等待线程 int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.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) compareAndSetWaitStatus(node, ws, 0); Node s = node.next; if (s == null || s.waitStatus > 0) { s = null; for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) LockSupport.unpark(s.thread); }
转载地址:http://wdgjx.baihongyu.com/