什么是AQS
1.java.util.concurrent包中的大多数同步器实现都是围绕着共同的基础行为,比如等待队列、条件队列、独占获取、共享获取等,而这些行为的抽象就是基于AbstractQueuedSynchronizer(简称AQS)实现的,AQS是一个抽象同步框架,可以用来实现一个依赖状态的同步器。
2.JDK中提供的大多数的同步器如Lock,Latch,Barrier等,都是基于AQS框架来实现的
一般是通过一个内部类Sync继承AQS
将同步器所有调用都映射到Sync对应的方法
AQS具备的特性:
1.阻塞等待队列,2.共享/独占,3.公平/非公平,4.可重入,5.允许中断
AQS定义两种资源共享方式
1.Exclusive-独占,只有一个线程能执行,如ReentrantLock(详情可查看深入理解ReentrantLock类锁)
2.Share-共享,多个线程可以同时执行,如Semaphore/CountDownLatch
AQS定义两种队列
1.同步等待队列
AQS当中的同步等待队列也称CLH队列,CLH队列是Craig、Landin、Hagersten三人发明的一种基于双向链表数据结构的队列,是FIFO先进先出线程等待队列,Java中的CLH队列是原CLH队列的一个变种,线程由原自旋机制改为阻塞机制。
AQS依赖CLH同步队列来完成同步状态的管理:
1)当前线程如果获取同步状态失败时,AQS则会将当前线程已经等待状态等信息构造成一个节点(Node)并将其加入到CLH同步队列,同时会阻塞当前线程
2)当同步状态释放时,会把首节点唤醒(公平锁),使其再次尝试获取同步状态。
3)通过signal或signalAll将条件队列中的节点转移到同步队列。(由条件队列转化为同步队列)
图示:
2.条件等待队列
AQS中条件队列是使用单向列表保存的,用nextWaiter来连接:
1)调用await方法阻塞线程;
2)当前线程存在于同步队列的头结点,调用await方法进行阻塞(从同步队列转化到条件队列)
3.AQS定义了5个队列中节点状态:
1)值为0,初始化状态,表示当前节点在sync队列中,等待着获取锁。
2)CANCELLED,值为1,表示当前的线程被取消;
3)SIGNAL,值为-1,表示当前节点的后继节点包含的线程需要运行,也就是unpark;
4)CONDITION,值为-2,表示当前节点在等待condition,也就是在condition队列中;
5)PROPAGATE,值为-3,表示当前场景下后续的acquireShared能够得以执行;
源码详解(将源码拆分为三块,抽象同步器AbstractQueuedSynchronizer类,节点Node类,条件对象ConditionObject类)
AbstractQueuedSynchronizer类解析
1.属性值解析
//用链表来表示队列privatetransientvolatileNodehead;privatetransientvolatileNodetail;privatevolatileintstate;//可以表示锁的加锁状态,又可以表示锁的重入次数,0为没有加锁
2.方法解析
//定义了主体的大体逻辑,如入队,如尝试加锁privateNodeaddWaiter(Nodemode){Nodenode=newNode(Thread.currentThread(),mode);//Trythefastpathofenq;backuptofullenqonfailureNodepred=tail;if(pred!=null){node.prev=pred;if(
Node类详解
1.代码展示
staticfinalclassNode{staticfinalNodeSHARED=newNode();//共享模式标记staticfinalNodeEXCLUSIVE=null;//独占模式标记staticfinalintCANCELLED=1;staticfinalintSIGNAL=-1;staticfinalintCONDITION=-2;staticfinalintPROPAGATE=-3;//值为0,初始化状态,表示当前节点在sync队列中,等待着获取锁。//CANCELLED,值为1,表示当前的线程被取消;//SIGNAL,值为-1,表示当前节点的后继节点包含的线程需要运行,也就是unpark;//CONDITION,值为-2,表示当前节点在等待condition,也就是在condition队列中;//PROPAGATE,值为-3,表示当前场景下后续的acquireShared能够得以执行;volatileintwaitStatus;volatileNodeprev;//前驱结点volatileNodenext;//后继结点volatileThreadthread;//与节点绑定的线程NodenextWaiter;//存储condition队列中的后继节点finalbooleanisShared(){returnnextWaiter==SHARED;}finalNodepredecessor()throwsNullPointerException{Nodep=prev;if(p==null)thrownewNullPointerException();elsereturnp;}Node(){}Node(Threadthread,Nodemode){//UsedbyaddWaiterthis.nextWaiter=mode;this.thread=thread;}Node(Threadthread,intwaitStatus){//UsedbyConditionthis.waitStatus=waitStatus;this.thread=thread;}}
Condition接口详解
1.代码展示
//Condition用来替代synchronized锁的监视器的功能,而且更加灵活//一个Condition实例需要与一个lock进行绑定publicinterfaceCondition{//调用此方法的线程将加入等待队列,阻塞直到被通知或者线程中断voidawait()throwsInterruptedException;//调用此方法的线程将加入等待队列,阻塞直到被通知(线程中断忽略)voidawaitUninterruptibly();//调用此方法的线程将加入等待队列,阻塞直到被通知或者线程中断或等待超时longawaitNanos(longnanosTimeout)throwsInterruptedException;//调用此方法的线程将加入等待队列,阻塞直到被通知或者线程中断或等待超时booleanawait(longtime,TimeUnitunit)throwsInterruptedException;//调用此方法的线程将加入等待队列,阻塞直到被通知或者线程中断或超出指定日期booleanawaitUntil(Datedeadline)throwsInterruptedException;//唤醒一个等待中的线程voidsignal();//唤醒所以等待中的线程voidsignalAll();}
2.发现说明
在Condition中,用await()替换wait(),用signal()替换notify(),用signalAll()替换notifyAll(),传统线程的通信方式,Condition都可以实现,这里注意,Condition是被绑定到Lock上的,要创建一个Lock的Condition必须用newCondition()方法。Condition的强大之处在于,对于一个锁,我们可以为多个线程间建立不同的Condition。如果采用Object类中的wait(),notify(),notifyAll()实现的话,当写入数据之后需要唤醒读线程时,不可能通过notify()或notifyAll()明确的指定唤醒读线程,而只能通过notifyAll唤醒所有线程,但是notifyAll无法区分唤醒的线程是读线程,还是写线程。所以,通过Condition能够更加精细的控制多线程的休眠与唤醒。
但,condition的使用必须依赖于lock对象,通过lock对象的newCondition()方法初始化一个condition对象。
ConditionObject类详解
1.属性值解析
//由头尾两个节点指针形成的链表来达到队列的效果privatetransientNodefirstWaiter;privatetransientNodelastWaiter;
2.方法解析
核心await方法
publicfinalvoidawait()throwsInterruptedException{//如果线程中断,直接抛出异常if(Thread.interrupted())thrownewInterruptedException();//进入等待队列中Nodenode=addConditionWaiter();//释放当前线程持有的锁,并获取当前同步器状态intsavedState=fullyRelease(node);intinterruptMode=0;//如果不在同步队列中,那么直接阻塞当前线程;直到被唤醒时,加入到同步队列中while(!isOnSyncQueue(node)){LockSupport.park(this);if((interruptMode=checkInterruptWhileWaiting(node))!=0)break;}//此时已经被唤醒,那么尝试获取锁if(acquireQueued(node,savedState)interruptMode!=THROW_IE)interruptMode=REINTERRUPT;//如果节点中断取消,那么清除节点if(node.nextWaiter!=null)//cleanupifcancelledunlinkCancelledWaiters();if(interruptMode!=0)reportInterruptAfterWait(interruptMode);}//addConditionWaiter将一个节点添加到condition队列中。在入队时,判断当前尾节点是不是CONDITION。如果不是则判断当前尾节点已经被取消,将当前节点出队。那么也就是说在队列中的节点状态,要么是CONDITION,要么是CANCELLEDprivateNodeaddConditionWaiter(){Nodet=lastWaiter;//IflastWaiteriscancelled,cleanout.if(t!=nullt.waitStatus!=Node.CONDITION){unlinkCancelledWaiters();t=lastWaiter;}Nodenode=newNode(Thread.currentThread(),Node.CONDITION);if(t==null)firstWaiter=node;elset.nextWaiter=node;lastWaiter=node;returnnode;}//方法的作用是移除取消的节点。方法本身只有在持有锁的时候会被调用。方法会遍历当前condition队列,将所有非Condition状态的节点移除。privatevoidunlinkCancelledWaiters(){Nodet=firstWaiter;Nodetrail=null;while(t!=null){Nodenext=t.nextWaiter;if(t.waitStatus!=Node.CONDITION){t.nextWaiter=null;if(trail==null)firstWaiter=next;elsetrail.nextWaiter=next;if(next==null)lastWaiter=trail;}elsetrail=t;t=next;}}
核心signal方法与signalAll方法
publicfinalvoidsignal(){if(!isHeldExclusively())thrownewIllegalMonitorStateException();Nodefirst=firstWaiter;if(first!=null)doSignal(first);}publicfinalvoidsignalAll(){if(!isHeldExclusively())thrownewIllegalMonitorStateException();Nodefirst=firstWaiter;if(first!=null)doSignalAll(first);}privatevoiddoSignal(Nodefirst){do{if((firstWaiter=first.nextWaiter)==null)lastWaiter=null;first.nextWaiter=null;}while(!transferForSignal(first)(first=firstWaiter)!=null);}privatevoiddoSignalAll(Nodefirst){lastWaiter=firstWaiter=null;do{Nodenext=first.nextWaiter;first.nextWaiter=null;transferForSignal(first);first=next;}while(first!=null);}finalbooleantransferForSignal(Nodenode){//如果不能更改waitStatus,则表示该节点已被取消if(!