一文详解 Java 的几把 JVM 级锁( 五 )_作者|楚昭在计算机行业有一个定律叫\

ReentrantReadWriteLock

ReentrantReadWriteLock (读写锁)其实是两把锁，一把是 WriteLock (写锁) ，一把是读锁， ReadLock 。读写锁的规则是：读读不互斥、读写互斥、写写互斥。在一些实际的场景中，读操作的频率远远高于写操作，如果直接用一般的锁进行并发控制的话，就会读读互斥、读写互斥、写写互斥，效率低下，读写锁的产生就是为了优化这种场景的操作效率。一般情况下独占锁的效率低来源于高并发下对临界区的激烈竞争导致线程上下文切换。因此当并发不是很高的情况下，读写锁由于需要额外维护读锁的状态，可能还不如独占锁的效率高，因此需要根据实际情况选择使用。 ReentrantReadWriteLock 的原理也是基于 AQS 进行实现的，与 ReentrantLock 的差别在于 ReentrantReadWriteLock 锁拥有共享锁、排他锁属性。读写锁中的加锁、释放锁也是基于 Sync (继承于 AQS ) ，并且主要使用 AQS 中的 state 和 node 中的 waitState 变量进行实现的。实现读写锁与实现普通互斥锁的主要区别在于需要分别记录读锁状态及写锁状态，并且等待队列中需要区别处理两种加锁操作。 ReentrantReadWriteLock 中将 AQS 中的 int 类型的 state 分为高 16 位与第 16 位分别记录读锁和写锁的状态，如下图所示： WriteLock(写锁)是悲观锁(排他锁、互斥锁) 通过计算 state((116)-1) ，将 state 的高 16 位全部抹去，因此 state 的低位记录着写锁的重入计数。获取写锁源码： /** *获取写锁 Acquiresthewritelock. *如果此时没有任何线程持有写锁或者读锁，那么当前线程执行CAS操作更新status ， *若更新成功，则设置读锁重入次数为1 ，并立即返回 *pAcquiresthewritelockifneitherthereadnorwritelock *areheldbyanotherthread *andreturnsimmediatelysettingthewritelockholdcountto *one. *如果当前线程已经持有该写锁，那么将写锁持有次数设置为1 ，并立即返回 *pIfthecurrentthreadalreadyholdsthewritelockthenthe *holdcountisincrementedbyoneandthemethodreturns *immediately. *如果该锁已经被另外一个线程持有，那么停止该线程的CPU调度并进入休眠状态， *直到该写锁被释放，且成功将写锁持有次数设置为1才表示获取写锁成功 *pIfthelockisheldbyanotherthreadthenthecurrent *threadbecomesdisabledforthreadschedulingpurposesand *liesdormantuntilthewritelockhasbeenacquiredatwhich *timethewritelockholdcountissettoone. */ public void lock () { sync.acquire( 1 ); /** *该方法为以独占模式获取锁，忽略中断 *如果调用一次该“tryAcquire”方法更新status成功，则直接返回，代表抢锁成功 *否则，将会进入同步队列等待，不断执行“tryAcquire”方法尝试CAS更新status状态，直到成功抢到锁 *其中“tryAcquire”方法在NonfairSync(公平锁)中和FairSync(非公平锁)中都有各自的实现 * *Acquiresinexclusivemodeignoringinterrupts.Implemented *byinvokingatleastonce{ @link #tryAcquire *returningonsuccess.Otherwisethethreadisqueuedpossibly *repeatedlyblockingandunblockinginvoking{ @link *#tryAcquireuntilsuccess.Thismethodcanbeused *toimplementmethod{ @link Lock#lock. * * @param argtheacquireargument.Thisvalueisconveyedto *{ @link #tryAcquirebutisotherwiseuninterpretedand *canrepresentanythingyoulike. */ public final void acquire ( int arg) { if (!tryAcquire(arg) acquireQueued(addWaiter(Node.EXCLUSIVE)arg)) selfInterrupt(); protected final boolean tryAcquire ( int acquires) { /* *Walkthrough: *1、如果读写锁的计数不为0 ，且持有锁的线程不是当前线程，则返回false *1.Ifreadcountnonzeroorwritecountnonzero *andownerisadifferentthreadfail. *2、如果持有锁的计数不为0且计数总数超过限定的最大值，也返回false *2.Ifcountwouldsaturatefail.(Thiscanonly *happenifcountisalreadynonzero.) *3、如果该锁是可重入或该线程在队列中的策略是允许它尝试抢锁，那么该线程就能获取锁 *3.Otherwisethisthreadiseligibleforlockif *itiseitherareentrantacquireor *queuepolicyallowsit.Ifsoupdatestate *andsetowner. */ Threadcurrent=Thread.currentThread(); //获取读写锁的状态 int c=getState(); //获取该写锁重入的次数 int w=exclusiveCount(c); //如果读写锁状态不为0 ，说明已经有其他线程获取了读锁或写锁 if (c!= 0 ){ //如果写锁重入次数为0 ，说明有线程获取到读锁，根据“读写锁互斥”原则，返回false //或者如果写锁重入次数不为0 ，且获取写锁的线程不是当前线程，根据\"写锁独占\"原则，返回false //(Note:ifc!=0andw==0thensharedcount!=0) if (w== 0 ||current!=getExclusiveOwnerThread()) return false ; //如果写锁可重入次数超过最大次数（65535），则抛异常 if (w+exclusiveCount(acquires)MAX_COUNT) throw new Error( \"Maximumlockcountexceeded\" ); //到这里说明该线程是重入写锁，更新重入写锁的计数(+1) ，返回true //Reentrantacquire setState(c+acquires); return true ; //如果读写锁状态为0说明读锁和写锁都没有被获取，会走下面两个分支： //如果要阻塞或者执行CAS操作更新读写锁的状态失败，则返回false //如果不需要阻塞且CAS操作成功，则当前线程成功拿到锁，设置锁的owner为当前线程，返回true if (writerShouldBlock()|| !compareAndSetState(cc+acquires)) return false ; setExclusiveOwnerThread(current); return true ; 释放写锁源码： /* *NotethattryReleaseandtryAcquirecanbecalledby *Conditions.Soitispossiblethattheirargumentscontain *bothreadandwriteholdsthatareallreleasedduringa *conditionwaitandre-establishedintryAcquire. */ protected final boolean tryRelease ( int releases) { //若锁的持有者不是当前线程，抛出异常 if (!isHeldExclusively()) throw new IllegalMonitorStateException(); //写锁的可重入计数减掉releases个 int nextc=getState()-releases; //如果写锁重入计数为0了，则说明写锁被释放了 boolean free=exclusiveCount(nextc)== 0 ; if (free) //若写锁被释放，则将锁的持有者设置为null ，进行GC setExclusiveOwnerThread( null ); //更新写锁的重入计数 setState(nextc); return free; ReadLock(读锁)是共享锁(乐观锁) 通过计算 state16 进行无符号补 0 ，右移 16 位，因此 state 的高位记录着写锁的重入计数. 读锁获取锁的过程比写锁稍微复杂些，首先判断写锁是否为 0 并且当前线程不占有独占锁，直接返回；否则，判断读线程是否需要被阻塞并且读锁数量是否小于最大值并且比较设置状态成功，若当前没有读锁，则设置第一个读线程 firstReader 和 firstReaderHoldCount ；若当前线程线程为第一个读线程，则增加 firstReaderHoldCount ；否则，将设置当前线程对应的 HoldCounter 对象的值，更新成功后会在 firstReaderHoldCount 中 readHolds ( ThreadLocal 类型的)的本线程副本中记录当前线程重入数，这是为了实现 JDK1.6 中加入的 getReadHoldCount ()方法的，这个方法能获取当前线程重入共享锁的次数( state 中记录的是多个线程的总重入次数) ，加入了这个方法让代码复杂了不少，但是其原理还是很简单的：如果当前只有一个线程的话，还不需要动用 ThreadLocal ，直接往 firstReaderHoldCount 这个成员变量里存重入数，当有第二个线程来的时候，就要动用 ThreadLocal 变量 readHolds 了，每个线程拥有自己的副本，用来保存自己的重入数。获取读锁源码： /** *获取读锁 *Acquiresthereadlock. *如果写锁未被其他线程持有，执行CAS操作更新status值，获取读锁后立即返回 *pAcquiresthereadlockifthewritelockisnotheldby *anotherthreadandreturnsimmediately. * *如果写锁被其他线程持有，那么停止该线程的CPU调度并进入休眠状态，直到该读锁被释放 *pIfthewritelockisheldbyanotherthreadthen *thecurrentthreadbecomesdisabledforthreadscheduling *purposesandliesdormantuntilthereadlockhasbeenacquired. */ public void lock () { sync.acquireShared( 1 ); /** *该方法为以共享模式获取读锁，忽略中断 *如果调用一次该“tryAcquireShared”方法更新status成功，则直接返回，代表抢锁成功 *否则，将会进入同步队列等待，不断执行“tryAcquireShared”方法尝试CAS更新status状态，直到成功抢到锁 *其中“tryAcquireShared”方法在NonfairSync(公平锁)中和FairSync(非公平锁)中都有各自的实现 *(看这注释是不是和写锁很对称) *Acquiresinsharedmodeignoringinterrupts.Implementedby *firstinvokingatleastonce{@link#tryAcquireShared *returningonsuccess.Otherwisethethreadisqueuedpossibly *repeatedlyblockingandunblockinginvoking{@link *#tryAcquireShareduntilsuccess. * *@paramargtheacquireargument.Thisvalueisconveyedto *{@link#tryAcquireSharedbutisotherwiseuninterpreted *andcanrepresentanythingyoulike. */ public final void acquireShared ( int arg ) { if (tryAcquireShared(arg) 0 ) doAcquireShared(arg); protected final int tryAcquireShared ( int unused ) { /* *Walkthrough: *1、如果已经有其他线程获取到了写锁，根据“读写互斥”原则，抢锁失败，返回-1 *1.Ifwritelockheldbyanotherthreadfail. *2、如果该线程本身持有写锁，那么看一下是否要readerShouldBlock ，如果不需要阻塞， *则执行CAS操作更新state和重入计数。 *这里要注意的是，上面的步骤不检查是否可重入(因为读锁属于共享锁，天生支持可重入) *2.Otherwisethisthreadiseligiblefor *lockwrtstatesoaskifitshouldblock *becauseofqueuepolicy.Ifnottry *tograntbyCASingstateandupdatingcount. *Notethatstepdoesnotcheckforreentrant *acquireswhichispostponedtofullversion *toavoidhavingtocheckholdcountin *themoretypicalnon-reentrantcase. *3、如果因为CAS更新status失败或者重入计数超过最大值导致步骤2执行失败 *那就进入到fullTryAcquireShared方法进行死循环，直到抢锁成功 *3.Ifstep2failseitherbecausethread *apparentlynoteligibleorCASfailsorcount *saturatedchaintoversionwithfullretryloop. */ //当前尝试获取读锁的线程 Threadcurrent=Thread.currentThread(); //获取该读写锁状态 int c=getState(); //如果有线程获取到了写锁，且获取写锁的不是当前线程则返回失败 if (exclusiveCount(c)!= 0 getExclusiveOwnerThread()!=current) return -1 ; //获取读锁的重入计数 int r=sharedCount(c); //如果读线程不应该被阻塞，且重入计数小于最大值，且CAS执行读锁重入计数+1成功，则执行线程重入的计数加1操作，返回成功 if (!readerShouldBlock() rMAX_COUNT compareAndSetState(cc+SHARED_UNIT)){ //如果还未有线程获取到读锁，则将firstReader设置为当前线程， firstReaderHoldCount设置为1 if (r== 0 ){ firstReader=current; firstReaderHoldCount= 1 ; else if (firstReader==current){ //如果firstReader是当前线程，则将firstReader的重入计数变量firstReaderHoldCount加1 firstReaderHoldCount++; else { //否则说明有至少两个线程共享读锁，获取共享锁重入计数器HoldCounter //从HoldCounter中拿到当前线程的线程变量cachedHoldCounter ，将此线程的重入计数count加1 HoldCounterrh=cachedHoldCounter; if (rh== null ||rh.tid!=getThreadId(current)) cachedHoldCounter=rh=readHolds. get (); else if (rh.count== 0 ) readHolds. set (rh); rh.count++; return 1 ; //如果上面的if条件有一个都不满足，则进入到这个方法里进行死循环重新获取 return fullTryAcquireShared(current); /** *用于处理CAS操作state失败和tryAcquireShared中未执行获取可重入锁动作的full方法(补偿方法？) *FullversionofacquireforreadsthathandlesCASmisses *andreentrantreadsnotdealtwithintryAcquireShared. */ final int fullTryAcquireShared ( Threadcurrent ) { /* *此代码与tryAcquireShared中的代码有部分相似的地方， *但总体上更简单，因为不会使tryAcquireShared与重试和延迟读取保持计数之间的复杂判断 *Thiscodeisinpartredundantwiththatin *tryAcquireSharedbutissimpleroverallbynot *complicatingtryAcquireSharedwithinteractionsbetween *retriesandlazilyreadingholdcounts. */ HoldCounterrh= null ; //死循环 for (;;){ //获取读写锁状态 int c=getState(); //如果有线程获取到了写锁 if (exclusiveCount(c)!= 0 ){ //如果获取写锁的线程不是当前线程，返回失败 if (getExclusiveOwnerThread()!=current) return -1 ; //elseweholdtheexclusivelock;blockinghere //wouldcausedeadlock. else if (readerShouldBlock()){ //如果没有线程获取到写锁，且读线程要阻塞 //Makesurewe'renotacquiringreadlockreentrantly //如果当前线程为第一个获取到读锁的线程 if (firstReader==current){ //assertfirstReaderHoldCount0; 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( \"Maximumlockcountexceeded\" ); //如果执行CAS操作成功将读写锁的重入计数加1 ，则对当前持有这个共享读锁的线程的重入计数加1 ，然后返回成功 if (compareAndSetState(cc+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; //cacheforrelease return 1 ; 释放读锁源码： /** *Releasesinsharedmode.Implementedbyunblockingoneormore *threadsif{ @link #tryReleaseSharedreturnstrue. * * @param argthereleaseargument.Thisvalueisconveyedto *{ @link #tryReleaseSharedbutisotherwiseuninterpreted *andcanrepresentanythingyoulike. * @return thevaluereturnedfrom{ @link #tryReleaseShared */ public final boolean releaseShared ( int arg) { if (tryReleaseShared(arg)){ //尝试释放一次共享锁计数 doReleaseShared(); //真正释放锁 return true ; return false ; /** *此方法表示读锁线程释放锁。 *首先判断当前线程是否为第一个读线程firstReader ， *若是，则判断第一个读线程占有的资源数firstReaderHoldCount是否为1 ，若是，则设置第一个读线程firstReader为空，否则，将第一个读线程占有的资源数firstReaderHoldCount减1；若当前线程不是第一个读线程，那么首先会获取缓存计数器（上一个读锁线程对应的计数器），若计数器为空或者tid不等于当前线程的tid值，则获取当前线程的计数器，如果计数器的计数count小于等于1 ，则移除当前线程对应的计数器，如果计数器的计数count小于等于0 ，则抛出异常，之后再减少计数即可。无论何种情况，都会进入死循环，该循环可以确保成功设置状态state */ protected final boolean tryReleaseShared ( int unused) { //获取当前线程 Threadcurrent=Thread.currentThread(); if (firstReader==current){ //当前线程为第一个读线程 //assertfirstReaderHoldCount0; if (firstReaderHoldCount== 1 ) //读线程占用的资源数为1 firstReader= null ; else //减少占用的资源 firstReaderHoldCount--; else { //当前线程不为第一个读线程 //获取缓存的计数器 HoldCounterrh=cachedHoldCounter; if (rh== null ||rh.tid!=getThreadId(current)) //计数器为空或者计数器的tid不为当前正在运行的线程的tid //获取当前线程对应的计数器 rh=readHolds.get(); //获取计数 int count=rh.count; if (count= 1 ){ //计数小于等于1 //移除 readHolds.remove(); if (count= 0 ) //计数小于等于0 ，抛出异常 throw unmatchedUnlockException(); //减少计数 --rh.count; for (;;){ //死循环 //获取状态 int c=getState(); //获取状态 int nextc=c-SHARED_UNIT; if (compareAndSetState(cnextc)) //比较并进行设置 //Releasingthereadlockhasnoeffectonreaders //butitmayallowwaitingwriterstoproceedif //bothreadandwritelocksarenowfree. return nextc== 0 ; /**真正释放锁 *Releaseactionforsharedmode--signalssuccessorandensures *propagation.(Note:Forexclusivemodereleasejustamounts *tocallingunparkSuccessorofheadifitneedssignal.) */ private void doReleaseShared () { /* *Ensurethatareleasepropagatesevenifthereareother *in-progressacquires/releases.Thisproceedsintheusual *wayoftryingtounparkSuccessorofheadifitneeds *signal.ButifitdoesnotstatusissettoPROPAGATEto *ensurethatuponreleasepropagationcontinues. *Additionallywemustloopincaseanewnodeisadded *whilewearedoingthis.Alsounlikeotherusesof *unparkSuccessorweneedtoknowifCAStoresetstatus *failsifsorechecking. */ for (;;){ Nodeh=head; if (h!= null h!=tail){ int ws=h.waitStatus; if (ws==Node.SIGNAL){ if (!compareAndSetWaitStatus(hNode.SIGNAL 0 )) continue ; //looptorecheckcases unparkSuccessor(h); else if (ws== 0 !compareAndSetWaitStatus(h 0 Node.PROPAGATE)) continue ; //looponfailedCAS if (h==head) //loopifheadchanged break ; 通过分析可以看出：在线程持有读锁的情况下，该线程不能取得写锁(因为获取写锁的时候，如果发现当前的读锁被占用，就马上获取失败，不管读锁是不是被当前线程持有) 。在线程持有写锁的情况下，该线程可以继续获取读锁（获取读锁时如果发现写锁被占用，只有写锁没有被当前线程占用的情况才会获取失败）。