ReetrantLock源码分析
ReentrantLock类的大部分逻辑，都是其均继承自AQS的内部类Sync实现的 啥是AQS: Java并发编程核心在于java.concurrent.util包而juc当中的大多数同步器实现都是围绕着共同的基础行为，比如「等待队列、条件队列、独占获取、共享获取」等，而这个行为的抽象就是基于AbstractQueuedSynchronizer简称AQS 它定义了一套多线程访问共享资源的同步器框架，是一个「依赖状态(state)的同步器」。 以公平锁为例子: publicstaticvoidmain(String[]args){ ReentrantLocklock=newReentrantLock(true); lock.lock();//加锁断点处 try{ Thread.sleep(5000); }catch(InterruptedExceptione){ e.printStackTrace(); }finally{ lock.unlock(); } 公平锁、非公平锁 publicReentrantLock(booleanfair){//ReetrantLock的有参构造函数 sync=fair?newFairSync():newNonfairSync(); } 在加锁行打断点运行, 我们可以看到参数: image-20201105163128535 记住这几个值,后面会用到. 单步步入 F7 image-20201105163427610 看到lock()调用了Sync实例中的lock()方法 我们可以看到Sync是在ReentrantLock类中的一个抽象内部类 继承于AbstractQueuedSynchronizer (AQS)(抽象队列同步器) 点击Sync类中lock抽象接口方法 我们可以发现ReetrantLock有公平和非公平锁两种方式实现方式.由于本章我们使用公平锁讲解所以我们选择公平锁的实现方式继续向下调试代码 /** *Syncobjectforfairlocks */ staticfinalclassFairSyncextendsSync{//公平锁实现方式继承于Sync privatestaticfinallongserialVersionUID=-3000897897090466540L; finalvoidlock(){ acquire(1);//获取锁传入参数1 } .. 向下⬇️ publicfinalvoidacquire(intarg){//acquire方法接收参数1 if(!tryAcquire(arg)&&acquireQueued(addWaiter(Node.EXCLUSIVE),arg)) //这里有两个操作使用短路与如果第一个操作成功第二个操作不执行 //1.尝试以独占模式获取锁,在成功时返回 //2.线程进入队列排队 selfInterrupt(); } 方式1 首先尝试获取锁 /** *FairversionoftryAcquire.Don'tgrantaccessunless *recursivecallornowaitersorisfirst. */ protectedfinalbooleantryAcquire(intacquires){//参数:1 finalThreadcurrent=Thread.currentThread();//获取当前线程 intc=getState();//还记得我们第一个断点截图的state值是0 if(c==0){//是0说明我们是第一次获取锁,如果不是0说明是重入,会进入else if(!hasQueuedPredecessors()&& //公平锁和非公平锁，主要是在方法tryAcquire中，是否有!hasQueuedPredecessors()判断。 compareAndSetState(0,acquires)){//通过cas操作state更新,期望值是0,新值是1 setExclusiveOwnerThread(current);//设置当前拥有独占访问的线程。 returntrue; } } elseif(current==getExclusiveOwnerThread()){//如果是重入情况会进入这里. //进入前会判断是否是当前线程拿的锁 intnextc=c+acquires;//假设重入次数为2那么nextc=2+1=3 if(nextc<0) thrownewError("Maximumlockcountexceeded"); setState(nextc);//更新state值 returntrue; } returnfalse; } 先判断state是否为0，「如果为0就执行上面提到的lock方法的前半部分」，通过CAS操作将state的值从0变为1，否则判断当前线程是否为exclusiveOwnerThread，然后把state++，也就是重入锁的体现，「我们注意前半部分是通过CAS来保证同步，后半部分并没有同步的体现」，原因是：后半部分是线程重入，再次获得锁时才触发的操作，此时当前线程拥有锁，所以对ReentrantLock的属性操作是无需加锁的。「如果tryAcquire()获取失败，则要执行addWaiter()向等待队列中添加一个独占模式的节点。」 publicfinalbooleanhasQueuedPredecessors(){ Nodet=tail;//根据初始化顺序倒序获取字段 Nodeh=head; Nodes; returnh!=t&& ((s=h.next)==null||s.thread!=Thread.currentThread()); } //在这个判断中主要就是看当前线程是不是同步队列的首位，是：true、否：false //这部分涉及公平锁的实现，CLH（Craig，LandinandHagersten）。三个作者的首字母组合 啥是CLH CLH锁即Craig, Landin, and Hagersten (CLH) locks。CLH锁是一个自旋锁。能确保无饥饿性。提供先来先服务的公平性。 为什么说JUC中的实现是基于CLH的“变种”，因为原始CLH队列，一般用于实现自旋锁。而JUC中的实现，获取不到锁的线程，一般会时而阻塞，时而唤醒。 1.获取不到锁的线程，会进入队尾，然后自旋，直到其前驱线程释放锁,具体位置是放在tail后的null位置,并让新对象的next指向Null 2.如果head成功拿到了锁 此时 把head中包含的线程指向为Null并将head的前任设置为head后清除现任head 方式2 没获取到锁,需要进入队列排队: 进入方式需要改造项目让多个线程进行抢占资源改造如下: importjava.util.concurrent.locks.ReentrantLock; classScratch{ publicstaticvoidmain(String[]args){ ReentrantLocklock=newReentrantLock(true); for(inti=0;i<5;i++){ newThread(()->{ lock.lock(); try{ Thread.sleep(100); System.out.println(Thread.currentThread().getName()+"工作结束...."); }catch(InterruptedExceptione){ e.printStackTrace(); }finally{ lock.unlock(); } }).start(); } } } 回到这里发现等待队列addWaiter方法添加了一个Null 将自己加入CLH队列的尾部 关注公众号:[JAVA宝典] image-20201105171529838 发现等待队列addWaiter方法添加了一个Null null的含义是: 使用null是私有占用资源模式,使用new Node()是 共享模式.(顺带一提,writelock不互斥,就是使用的共享模式) Node还有几个等待状态: /** *Statusfield,takingononlythevalues: *SIGNAL:Thesuccessorofthisnodeis(orwillsoonbe) *blocked(viapark),sothecurrentnodemust *unparkitssuccessorwhenitreleasesor *cancels.Toavoidraces,acquiremethodsmust *firstindicatetheyneedasignal, *thenretrytheatomicacquire,andthen, *onfailure,block. *CANCELLED:Thisnodeiscancelledduetotimeoutorinterrupt. *Nodesneverleavethisstate.Inparticular, *athreadwithcancellednodeneveragainblocks. *CONDITION:Thisnodeiscurrentlyonaconditionqueue. *Itwillnotbeusedasasyncqueuenode *untiltransferred,atwhichtimethestatus *willbesetto0.(Useofthisvalueherehas *nothingtodowiththeotherusesofthe *field,butsimplifiesmechanics.) *PROPAGATE:AreleaseSharedshouldbepropagatedtoother *nodes.Thisisset(forheadnodeonly)in *doReleaseSharedtoensurepropagation *continues,evenifotheroperationshave *sinceintervened. *0:Noneoftheabove * *Thevaluesarearrangednumericallytosimplifyuse. *Non-negativevaluesmeanthatanodedoesn'tneedto *signal.So,mostcodedoesn'tneedtocheckforparticular *values,justforsign. * *Thefieldisinitializedto0fornormalsyncnodes,and *CONDITIONforconditionnodes.ItismodifiedusingCAS *(orwhenpossible,unconditionalvolatilewrites). */ volatileintwaitStatus; 在addWaiter(Node.EXCLUSIVE)处断点: /** *根据给定的模式为当前节点创建一个Node * *@parammodeNode.EXCLUSIVEforexclusive,Node.SHAREDforshared // *@returnthenewnode */ privateNodeaddWaiter(Nodemode){ //线程对应的Node Nodenode=newNode(Thread.currentThread(),mode); //Trythefastpathofenq;backuptofullenqonfailure Nodepred=tail; //尾节点不为空 if(pred!=null){ //当前node的前驱指向尾节点 node.prev=pred; //将当前node设置为新的尾节点 //如果cas操作失败，说明线程竞争 if(compareAndSetTail(pred,node)){ pred.next=node; returnnode; } } //lockfree的方式插入队尾 enq(node);//只有在tail==null时才进入 returnnode; } 先找到等待队列的tail节点pred，如果pred！=null，就把当前线程添加到pred后面进入等待队列，如果不存在tail节点执行enq() privateNodeenq(finalNodenode){ //经典的lockfree算法：循环+CAS for(;;){ Nodet=tail; //尾节点为空 if(t==null){//Mustinitialize //初始化头节点 if(compareAndSetHead(newNode())) tail=head; }else{ node.prev=t; if(compareAndSetTail(t,node)){ t.next=node; returnt; } } } } 这里进行了循环，「如果此时存在了tail就执行同上一步骤的添加队尾操作，如果依然不存在，就把当前线程作为head结点。」 插入节点后，调用acquireQueued()进行阻塞 /** *Acquiresinexclusiveuninterruptiblemodeforthreadalreadyin *queue.Usedbyconditionwaitmethodsaswellasacquire. * *@paramnodethenode *@paramargtheacquireargument *@return{@codetrue}ifinterruptedwhilewaiting */ finalbooleanacquireQueued(finalNodenode,intarg){ booleanfailed=true; try{ booleaninterrupted=false; for(;;){ finalNodep=node.predecessor(); if(p==head&&tryAcquire(arg)){ setHead(node); p.next=null;//helpGC failed=false; returninterrupted; } if(shouldParkAfterFailedAcquire(p,node)&& parkAndCheckInterrupt()) interrupted=true; } }finally{ if(failed) cancelAcquire(node); } } 先获取当前节点的前一节点p，如果p是head的话就再进行一次tryAcquire(arg)操作，如果成功就返回，否则就执行「shouldParkAfterFailedAcquire、parkAndCheckInterrupt来达到阻塞效果；」 unlock 在unlock处打断点 进入了 publicvoidunlock(){ sync.release(1); } image-20201106142501425 继续跟进发现进入release方法,继续查看tryRelease(arg)方法 尝试释放锁有三种实现 我们点ReetrantLock实现方式 源码: protectedfinalbooleantryRelease(intreleases){//最后进入到实际解锁源码中 intc=getState()-releases; if(Thread.currentThread()!=getExclusiveOwnerThread())//判断持有线程和当前执行线程是否是同一个,否则报错 thrownewIllegalMonitorStateException(); booleanfree=false; if(c==0){//判断释放掉releases参数后是否是0如果为0设置当前排他锁的占有线程为Null free=true; setExclusiveOwnerThread(null); } setState(c);//更新state为0 returnfree;//返回解锁是否成功 }
