一、ReentrantReadWriteLock简介 　　

　　今天，我们讨论的是可重入读写锁，它支持多个线程同时获取锁。然而，当获得写锁时，其他写锁和读锁将阻塞；其实可以看到读写锁维护的是一对锁和一个写锁，其实就是一个排他锁、一个读锁和一个共享锁。通过将读锁和写锁分开，与一般的排他锁相比，并发性大大提高。读写锁的性能优于排他锁，因为大多数场景下读大于写；读写锁提供公平的选择和重新进入(锁支持重新进入。以读写锁线程为例：获取读锁后，读线程可以再次获取读锁。写线程可以在获取写锁之后再次获取写锁，或者同时获取读锁)和锁降级(根据释放写锁之前获取写锁和读锁的顺序，可以将写锁降级为读锁)。 　　

　　二、ReentrantReadWriteLock基本成员 　　

　　我们先来看一个可重入的读写锁类图。 　　

　　图像处理器/2d 191001-85 E4-4 f18-bbfc-00656923 Fe 55/ 　　

　　Sync是继承AQS的内部类，主要实现AQS的一些方法。 　　

　　图像处理器/ea 467173-bd58-47a 7-9d6c-0115 b 883 aff 0/ 　　

　　基本成员介绍 　　

　　final int SHARED _ SHIFT=16//这是读取锁的原始累加值(即每次获取读取锁时，都会获取状态，然后添加到其中)，为2.16//例如假设当前状态为1，那么现在获取读取锁为1 SHARED _ unit static final int SHARED _ unit=(1 SHARED _ SHIFT)；//读锁和写锁的最大数量为2 16-1静态最终int max _ count=(1 shared _ shift)-1；//写锁的掩码实际上是2 16-1。这个数字的二进制非常特殊。所有16位都是1。静态最终int exclusive _ mask=(1 shared _ shift)-1。/ Returns the number of shared holds represented in count */ // 读锁的数量 static int sharedCount(int c) { return c SHARED_SHIFT; } /返回以count *表示的独占保持数////写锁定静态int独占计数(int c){ return c exclusive _ mask；}//记录每个线程获取的读锁数量。//count是一个数字。//tid是线程的唯一标识符。最终类hold counter { int count=0；//使用id而不是引用，以避免垃圾保留。最终长度tid=GetThreadID(thread . currentthread())；}//Inherit ThreadLocal，主要用于存储holdcounter静态最终类threadlocalholdcounter扩展thread local { public java . util . concurrent . locks . recentredwritellock . sync . hold cou . ente . r初始值(){返回新的Java。乌提尔。并发。锁。recentranreadwritellock。同步。保持计数器()；} }//可以理解为获取读锁的最后一个线程的私有瞬态保持计数器缓存holder//(用于优化性能，无需在ThreadLocal中查找)//获取读锁的第一个线程和读锁的数量，这意味着如果是第一个线程，则线程本地私有瞬态线程的第一个读取器=null私有瞬态intfirstflowholdcount。 　　

　　解释下读写锁的状态计算，state一个数，怎么控制的读和写 　　

　　我们先来看看写锁。让我们看看EXCLUSIVE_MASK= 　　

　　(1 SHARED_SHIFT) - 　　

　　1，这个数是65535，二进制数是161。我们看到获得写锁的数字C。 　　

　　65535(exclusiveCount方法，c代表状态，锁的状态，如果你不懂，可以看看AQS对状态的定义)。然而，由于二进制的16位是1，只要C在0-65535的范围内，它仍然是C(由于这个掩码的特殊性)，根据写锁的定义，只有 　　

这个二进制是不是等于0就可以了，所以了写锁的范围其实就是0-65535，其实二进制的范围就是低16位（因为最大数量是MAX_COUNT = 2^16-1）。

再来分析下读锁，读锁我们主要关注下SHARED_UNIT就可以，获取锁其实是用c+SHARED_UNIT（c代表的就是sate），释放锁是用c-

SHARED_UNIT，这个数是 65536（SHARED_UNIT 其实就是 2 ^

16），我们每次获取读锁其实就是把SHARED_UNIT累加，其实我们可以把SHARED_UNIT的一次累加就当做一次读锁的获取，我们看下读锁的值得范围是0

负65536(负数和int的最大值有关，int正数的最大值是2147483647，在给它累加其实就会变为负数，最后最大其实就是高16位全是1，因为读锁的最大数量是MAX_COUNT

= 2^16-1，所以其实读锁的范围可以看做是高16位)，获取读锁的个数就是无符号右移16位（就是sharedCount方法），因为可能是负数。

imagehandler/107a2bc5-7159-4702-86f0-55aec2087467'> 　　　　

上面的图片其实就是一个读锁，一个写锁，这种情况只有在同一个线程才会发生，如果你能算出一个写锁和读锁，说明你基本理解了读写锁状态控制的运算方法。

ReadLock和WriteLock，内部类，继承Lock，提供一些锁的方法。

FairSync和NonfairSync，内部类，是继承Sync，主要实现公平和非公平的一些方法

三、ReentrantReadWriteLock基本方法

1）、构造方法

// 无参，默认非公平public ReentrantReadWriteLock() { this(false); } // 有参 public ReentrantReadWriteLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); readerLock = new ReadLock(this); writerLock = new WriteLock(this); }

2）、WriteLock：写锁一些方法,如下图：

imagehandler/08c3f467-369d-4edf-8c96-b141c23bf364'> 　　　　

①、获取锁lock方法，我们可以看出调用的是sync.acquire方法，由于sync继承自AQS所以调用的其实是AQS的acquire，但是AQS的tryAcquire需要子类自己实现，所以我们看看tryAcquire（可以看出是个独占方法，也符合写锁的定义，只会有一个线程获取）。

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

子类sync的tryAcquire

// 写锁的状态控制（state） protected final boolean tryAcquire(int acquires) { /* * Walkthrough: * 1. If read count nonzero or write count nonzero * and owner is a different thread, fail. * 2. If count would saturate, fail. (This can only * happen if count is already nonzero.) * 3. Otherwise, this thread is eligible for lock if * it is either a reentrant acquire or * queue policy allows it. If so, update state * and set owner. */ // 获取当前线程 Thread current = Thread.currentThread(); // 获取当前锁的状态 int c = getState(); // 获取写锁的状态，c & （2的16次方-1） // 2的16次方-1 其实就是65535，变成二进制就是16个1 int w = exclusiveCount(c); // c不等于0，证明有线程获取锁了，不管是读锁或者写锁 if (c != 0) { // (Note: if c != 0 and w == 0 then shared count != 0) // w == 0，说明有读锁了，w!= 0证明有写锁 // current != getExclusiveOwnerThread()说明有写锁了，不是自己 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; } // 走到这一步，证明还没有线程获取锁 // writerShouldBlock 现在公平还是非公平，由FairSync和NonfairSync实现这个方法 if (writerShouldBlock() || // cas设置所得状态 !compareAndSetState(c, c + acquires)) //失败或者公平锁在我的前面有排队节点 return false; // 设置拥有锁的线程 setExclusiveOwnerThread(current); return true; }

在上面的写锁获取锁时我们需要注意下writerShouldBlock这个方法，它是实现公平和非公平的关键，它的公平和非公平的方法实现不同，公平是需要确认自己前面是否有排队节点，非公平直接返回false，具体查看这个方法。

②、写锁释放锁：unlock方法，它是其实也是调用的也是调用AQS的release方法，我们直接看子类的实现吧。

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

子类sync的tryRelease方法

protected final boolean tryRelease(int releases) { // 判断是否获取锁的是这个线程 if (!isHeldExclusively()) throw new IllegalMonitorStateException(); // 释放锁，修改state int nextc = getState() - releases; // 判断写锁是否完全释放 boolean free = exclusiveCount(nextc) == 0; // 完全释放，修改锁的拥有者为null if (free) setExclusiveOwnerThread(null); // cas状态 setState(nextc); return free; }

③、我们发现WriteLock里面还有一些获取锁的方法，lockInterruptibly响应中断，tryLock(long timeout,

TimeUnit

unit)支持中断并且带超时时间，其实都是调用了AQS里面的这些方法，然后获取锁这部分的实现都是调用的子类sync的tryAcquire方法。

3）、ReadLock：读锁的一些方法，如下图

imagehandler/80e941f8-8c7e-46bc-9aca-0f877b55fa77'> 　　　　

①、获取锁lock方法，调用过程和写锁一样，先走AQS,不过这一次调用的是共享锁的方法acquireShared，然后AQS在调用子类sync的实现方法。

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

sync的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(); // exclusiveCount(c) != 0 有线程获取了写锁 // 并且这个获取写锁的线程不是自己 if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1; // 现在只会有三种清理 // 1、还没有任何线程获取锁，不管是读锁还是写锁 // 2、有线程获取到了读锁 // 3、获取写锁的线程时自己 int r = sharedCount(c); // readerShouldBlock 由FairSync和NonfairSync实现 // FairSync 判断是否前面有排队节点 // NonfairSync排队节点是否有写节点 if (!readerShouldBlock() && r < max_count="" &&="" compareandsetstate(c,="" c="" +="" shared_unit))="" {="" r="=" 0证明还没有获取到锁="" if="" (r="=" 0)="" {="" firstreader="current;" firstreaderholdcount="1;" 重入锁="" }="" else="" if="" (firstreader="=" current)="" {="" firstreaderholdcount++;="" }="" else="" {="" 获取最后一个获取锁的holdcounter="" java.util.concurrent.locks.reentrantreadwritelock.sync.holdcounter="" rh="cachedHoldCounter;" 最后一个holdcounter是空或者不是本线程，就设置一个="" if="" (rh="=" null="" ||="" rh.tid="" !="getThreadId(current))" 其实这一步做了两件事情，其实是先set了一个holdcounter，然后在get之后设置给rh="" cachedholdcounter="rh" =="" readholds.get();="" else="" if="" (rh.count="=" 0)="" 理解不了什么时候会是0="" readholds.set(rh);="" rh.count++;="" }="" return="" 1;="" }="" 执行fulltryacquireshared方法有几种情况了="" 1.获取锁的下一个节点还是写锁，需要等待="" 2.到达获取锁的最大数量="" 3.可能存在多线程进程来设置读锁，cas失败了="" return="" fulltryacquireshared(current);="" }="" 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.="" */="" java.util.concurrent.locks.reentrantreadwritelock.sync.holdcounter="" rh="null;" for="" (;;)="" {="" 自旋获取锁="" int="" c="getState();" 获取锁状态="" if="" (exclusivecount(c)="" !="0)" {="" 判断有没有写锁，不等于0证明有写锁="" if="" (getexclusiveownerthread()="" !="current)" 写锁不是自己="" return="" -1;="" 返回="" 写锁时自己，其实就可以获取，其实就是锁降级（可以看做是一种特殊的重入锁）="" else="" we="" hold="" the="" exclusive="" lock;="" blocking="" here="" would="" cause="" deadlock.="" 到下面这个else="" if证明没有写锁="" readershouldblock="" 由fairsync和nonfairsync实现公平和非公平原则="" fairsync="" 判断是否前面有排队节点="" nonfairsync排队节点是否有写节点="" }="" else="" if="" (readershouldblock())="" {="" 到这一步证明了，是公平锁或者非公平锁的头结点.next是写锁,="" 此线程需要进入同步队列了,下面就是判断这个线程有没有获取过锁="" make="" sure="" we're="" not="" acquiring="" read="" lock="" reentrantly="" 第一个获取锁的是当前线程，证明可以重入="" if="" (firstreader="=" current)="" {="" assert="" firstreaderholdcount=""> 0; } else { // 进去这里说明，firstReader不是当前线程，那就说明获取读锁的不止一个了，因为firstReader不可能为null // 获取最后一个获取读锁的HoldCounter if (rh == null) { // rh == null 只会是第一次循环 rh = cachedHoldCounter; // 获取缓存的HoldCounter if (rh == null || rh.tid != getThreadId(current)) { // 从 ThreadLocal 中取出计数器，如果没有就会重新创建并设置 rh = readHolds.get(); if (rh.count == 0) // 那就证明没有获取到读读锁 readHolds.remove(); // 删除这个 } } if (rh.count == 0) // 这个是上面刚刚创建的证明获取锁失败了，需要进入队列 return -1; } } // 获取读锁的数量==MAX_COUNT，抛异常 if (sharedCount(c) == MAX_COUNT) throw new Error("Maximum lock count exceeded"); // 使用cas设置读锁的状态，下面逻辑和tryAcquireShared的逻辑一样 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; } } }

在读锁获取锁时需要注意下readerShouldBlock这个方法，和写锁类似，这个方法也是主要实现公平与非公平的关键，非公平锁（NonfairSync）时需要注意，如果获取读锁时，需要执行apparentlyFirstQueuedIsExclusive这个方法，判断队列head的next是否是写锁（是写锁，让这个写锁先来，避免写锁饥饿，就是避免写线程获取不到锁，所以写有很高的优先权），则自己获取读锁就需要排队，公平锁（FairSync）实现则还是需要判断队列里面是否有节点在排队。

②、释放锁unlock,调用AQS的releaseShared方法，我们主要关注子类实现

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

子类sync的tryReleaseShared方法

protected final boolean tryReleaseShared(int unused) { // 获取当前线程 Thread current = Thread.currentThread(); // 如果第一个读锁拥有者是当前线程 if (firstReader == current) { // assert firstReaderHoldCount > 0; // 读锁的数量为1，没有重入 if (firstReaderHoldCount == 1) firstReader = null; else // 重入了，修改数量 firstReaderHoldCount--; // 读锁已经不是一个了 } else { // 获取最后一个缓存 java.util.concurrent.locks.ReentrantReadWriteLock.Sync.HoldCounter rh = cachedHoldCounter; // 获取缓存HoldCounter if (rh == null || rh.tid != getThreadId(current)) // 不是就获取 rh = readHolds.get(); // 获取读锁线程的数量 int count = rh.count; // 如果小于等于1 if (count <= 1)="" {="" 删除这个线程的holdcounter="" readholds.remove();="" if="" (count=""><= 0)="" throw="" unmatchedunlockexception();="" }="" 读锁数量递减="" --rh.count;="" }="" for="" (;;)="" {="" 获取状态stase="" int="" c="getState();" 读锁状态减1(其实就是减shared_unit)="" int="" nextc="c" -="" shared_unit;="" cas设置线程状态="" 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.="" 为什么nextc="=" 0才会返回true了="" nextc="=" 0代表了什么了，没有读锁了="" 返回true，就代表可以去唤醒下一个线程了，但是队列的线程是写线程或者线程了，不确定="" 但是这个对读线程是没有影响的，但是对写锁是有影响的，我们想象一下什么情况下才会下个节点会是写锁获取的线程了="" 其实就是已经有线程获取了写锁，因为有线程获取了写锁，所以可能发生锁降级，写锁降级为读锁="" 为了保护锁降级的语义，所以必须保护读锁，直到没有读锁了才会去唤醒后面可能的写锁，也就是返回true="" return="" nextc="=" 0;="" }="">

③、lockInterruptibly和tryLock(long timeout, TimeUnit unit)和写锁的这些方法作用一样。

四、总结

> ReentrantReadWriteLock读写锁，要想学习好这个类，就必须了解什么是读锁，什么是写锁，怎么区别读锁和写锁，因为我们都知道锁都是通过AQS的state来控制的，但是现在是两个锁，所以理解ReentrantReadWriteLock对state拆分的运算很重要，也就是二进制的低16位是写锁，高16位是读锁，也不得不说大神设计让人叹为观止啊；还有就是理解读写锁的一些特性，重入指的就是同一个线程获取锁之后，再次调用lock方法不会被阻塞，但是注意调用几次lock，就要调用几次unlock，因为我们通过源码得知重入其实就是对state的累加，还有就是锁降级，我个人更愿意理解为特殊的重入，锁降级就是一个线程先获得写锁，然后这个线程再去获取读锁，这样不会阻塞，然后写锁其实就降级为了读锁，然后在释放写锁，最后释放读锁，作者为什么这么设计了，书上说的，锁降级主要是为了保证数据的可见性，如果当前线程不获取读锁而是直接释放写锁，假设此刻另一个线程（记作线程T）获取了写锁并修改了数据，那么当前线程无法感知线程T的数据更新，如果当前线程获取读锁，即遵循锁降级的步骤，则线程T,就会阻塞，直到当前线程使用数据并释放读锁之后，线程T才会获得写锁并更新数据。个人理解这个锁降级其实是一种特殊的锁的优化策略，在我们需要在边写边读的这种业务场景中，保证数据可见性的同时（不让其他线程获取写锁），提升本线程读锁性能，因为不需要和写锁或者其他读锁（公平锁）去竞争获取锁，而是直接降级为读锁。

> 参考 《Java 并发编程的艺术》