Java并发 结合源码分析AQS原理
前言：
如果说J.U.C包下的核心是什么？那我想答案只有一个就是AQS。那么AQS是什么呢？接下来让我们一起揭开AQS的神秘面纱
AQS是什么？
AQS是AbstractQueuedSynchronizer的简称。为什么说它是核心呢？是因为它提供了一个基于FIFO的队列和state变量来构建锁和其他同步装置的基础框架。下面是其底层的数据结构。
AQS的特点
1、其内使用Node实现FIFO（FirstInFirstOut）队列。可用于构建锁或者其他同步装置的基础框架
2、且利用了一个int类表示状态。在AQS中维护了一个volatileintstate，通常表示有线程访问资源的状态，当state>1的时候表示线程重入的数量，主要有三个方法控制：getState()，setState()，CompareAndSetState()。后面的源码分析多用到这几个方法
3、使用方法是继承，子类通过继承并通过实现它的方法管理其状态（acquire和release）的方法操纵状态。
4、同时实现排它锁和共享锁模式。实际上AQS功能主要分为两类：独占（只有一个线程能执行）和共享（多个线程同时执行），它的子类要么使用独占功能要么使用共享功能，而ReentrantLock是通过两个内部类来实现独占和共享
CountDownLatch如何借助AQS实现计数功能？
先来说一下CountDownLatch，CountDownLatch是一个同步辅助类，通过它可以来完成类似阻塞当前线程的功能，即一个或多个线程一起等待，直到其他线程执行的操作完成。要实现上面的功能，CountDownLatch是通过一个给定的原子操作的计数器来实现。调用该类的await()方法的线程会一直处于阻塞状态，直到其他线程调用countDown()方法使得计数器的值变为0之后线程才会执行，这个计数器是不能被重置的。通常这个类会用在程序执行需要等待某个条件完成的场景，比如说并行计算，可将一个数据量很大的计算拆分成一个个子任务，当子任务完成之后，再将最终的结果汇总。每次访问CountDownLatch只能有一个线程，但是这个线程在使用完countDown()方法之后能多个线程能继续运行，而调用await()方法的线程就一定要计数器为0才会运行
下面来分析CountDownLatch的源码以及如何使用AQS框架
publicclassCountDownLatch{
/**
*CountDownLatch实现同步控制
*底层是使用AQS的state来代表count
*/
privatestaticfinalclassSyncextendsAbstractQueuedSynchronizer{
privatestaticfinallongserialVersionUID=4982264981922014374L;
//初始化内部类实际上是设置AQS的state
Sync(intcount){
setState(count);
}
intgetCount(){
returngetState();
}
//尝试获取共享是看当前的state是否为0
protectedinttryAcquireShared(intacquires){
return(getState()==0)?1:-1;
}
/*尝试释放共享锁则是递减计数直到state==0就返回false代表资源已经释放完全否则就会使用CAS来让state减一*/
protectedbooleantryReleaseShared(intreleases){
for(;;){
intc=getState();
if(c==0)
returnfalse;
intnextc=c-1;
if(compareAndSetState(c,nextc))
returnnextc==0;
}
}
}
privatefinalSyncsync;
/**
*初始化CountDownLatch，实际上是初始化内部类，实际上是设置AQS的state，count不能小于0
*/
publicCountDownLatch(intcount){
if(count<0)thrownewIllegalArgumentException("count<0");
this.sync=newSync(count);
}
/**
*这里实际上是调用了AQS里的acquireSharedInterruptibly方法，完成的功能就是先去查看线程是否被中断，中断则抛出异常，没有被中断就会尝试获取共享资源。*注意在syn内部类中重写了tryAcquireShared，也就是当state为0就返回1，这时候就会将当前线程放入AQS的队列中去，也就是这时候线程可以不再阻塞而是尝试去获取锁
*/
publicvoidawait()throwsInterruptedException{
sync.acquireSharedInterruptibly(1);
}
/**
*原理同上面方法，但是加了一个时间参数来设置等待的时间
*/
publicbooleanawait(longtimeout,TimeUnitunit)
throwsInterruptedException{
returnsync.tryAcquireSharedNanos(1,unit.toNanos(timeout));
}
/**
*这里传入参数为1，同样上面内部类一样重写了AQS的tryReleaseShared方法，使用这个重写的方法来让计数器原子操作的减一
*/
publicvoidcountDown(){
sync.releaseShared(1);
}
/**
*就是获取AQS的state
*/
publiclonggetCount(){
returnsync.getCount();
}
/**
*转换成字符串的方法
*/
publicStringtoString(){
returnsuper.toString()+"[Count="+sync.getCount()+"]";
}
}
由上面代码可看见CountDownLatch实现了AQS的共享锁，原理是操作state来实现计数，并且重写了tryAcquireShared()，tryReleaseShared()等方法
Semaphore是如何借助AQS实现控制并发访问线程个数？
Semaphore的功能类似于操作系统的信号量，可以很方便的控制某个资源同时被几个线程访问，即做并发访问控制，与CountDownLatch类似，同样是实现获取和释放两个方法。Semaphore的使用场景：常用于仅能提供访问的资源，比如数据库的连接数最大只有30，而应用程序的并发数可能远远大于30，这时候就可以使用Semaphore来控制同时访问的线程数。当Semaphore控制线程数到1的时候就和我们单线程一样了。同样Semaphore说是信号量的意思，我们这里就可以把它理解为十字路口的红绿灯，可以控制车流量（这里是控制线程数）
下面来分析Semaphore的源码以及如何使用AQS框
publicclassSemaphoreimplementsjava.io.Serializable{privatestaticfinallongserialVersionUID=-3222578661600680210L;/**所有机制都通过AbstractQueuedSynchronizer子类实现*/
privatefinalSyncsync;
/**
*同样是通过内部类来实现AQS主要功能，使用state来表示许可证数量
*/
abstractstaticclassSyncextendsAbstractQueuedSynchronizer{
privatestaticfinallongserialVersionUID=1192457210091910933L;
Sync(intpermits){
setState(permits);
}
finalintgetPermits(){
returngetState();
}
/*
*不公平的获取方式，会有一个抢占锁的情况，即线程执行顺序会乱
*/
finalintnonfairTryAcquireShared(intacquires){
for(;;){
intavailable=getState();
intremaining=available-acquires;
if(remaining<0||
compareAndSetState(available,remaining))
returnremaining;
}
}
/*
*释放资源
*/
protectedfinalbooleantryReleaseShared(intreleases){
for(;;){
intcurrent=getState();
intnext=current+releases;
if(nextcurrent)//underflow
thrownewError("Permitcountunderflow");
if(compareAndSetState(current,next))
return;
}
}
finalintdrainPermits(){
for(;;){
intcurrent=getState();
if(current==0||compareAndSetState(current,0))
returncurrent;
}
}
}
/**
*不公平的sync版本，使用的就是sync定义的不公平锁
*/
staticfinalclassNonfairSyncextendsSync{
privatestaticfinallongserialVersionUID=-2694183684443567898L;
NonfairSync(intpermits){
super(permits);
}
protectedinttryAcquireShared(intacquires){
returnnonfairTryAcquireShared(acquires);
}
}
/**
*公平版本，获取锁的线程顺序就是线程启动的顺序。具体是使用hasQueuedPredecessors()方法判断“当前线程”是不是CLH队列中的第一个线程。*若不是的话，则返回-1，是就设置获取许可证，并检查许可证数量是否足够
*/
staticfinalclassFairSyncextendsSync{
privatestaticfinallongserialVersionUID=2014338818796000944L;
FairSync(intpermits){
super(permits);
}
protectedinttryAcquireShared(intacquires){
for(;;){
if(hasQueuedPredecessors())
return-1;
intavailable=getState();
intremaining=available-acquires;
if(remaining<0||
compareAndSetState(available,remaining))
returnremaining;
}
}
}
/**
*默认使用不公平的版本，如果需要公平的，则需要两个参数
*/
publicSemaphore(intpermits){
sync=newNonfairSync(permits);
}
publicSemaphore(intpermits,booleanfair){
sync=fair?newFairSync(permits):newNonfairSync(permits);
}
/**
*分析同CountDownLatch中的类似方法，具体的实现都是内部类中的获取方法，这里是获取一个许可
*/
publicvoidacquire()throwsInterruptedException{
sync.acquireSharedInterruptibly(1);
}
/**
*功能同上，但是这里不会检测线程是否被中断
*/
publicvoidacquireUninterruptibly(){
sync.acquireShared(1);
}
/**
*尝试获取
*/
publicbooleantryAcquire(){
returnsync.nonfairTryAcquireShared(1)>=0;
}
/**
*在一段时间内一直尝试获取许可
*/
publicbooleantryAcquire(longtimeout,TimeUnitunit)
throwsInterruptedException{
returnsync.tryAcquireSharedNanos(1,unit.toNanos(timeout));
}
/**
*当前线程释放一个许可证
*/
publicvoidrelease(){
sync.releaseShared(1);
}
/**
*可以规定一个线程获得许可证的数量
*/
publicvoidacquire(intpermits)throwsInterruptedException{
if(permits<0)thrownewIllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
publicvoidacquireUninterruptibly(intpermits){
if(permits<0)thrownewIllegalArgumentException();
sync.acquireShared(permits);
}
publicbooleantryAcquire(intpermits){
if(permits<0)thrownewIllegalArgumentException();
returnsync.nonfairTryAcquireShared(permits)>=0;
}
publicbooleantryAcquire(intpermits,longtimeout,TimeUnitunit)
throwsInterruptedException{
if(permits<0)thrownewIllegalArgumentException();
returnsync.tryAcquireSharedNanos(permits,unit.toNanos(timeout));
}
/**
*同样可以规定一个线程释放许可证的数量
*/
publicvoidrelease(intpermits){
if(permits<0)thrownewIllegalArgumentException();
sync.releaseShared(permits);
}
/**
*当前的许可还剩几个
*/
publicintavailablePermits(){
returnsync.getPermits();
}
/**
*销毁所有许可
*/
publicintdrainPermits(){
returnsync.drainPermits();
}
protectedvoidreducePermits(intreduction){
if(reduction<0)thrownewIllegalArgumentException();
sync.reducePermits(reduction);
}
publicfinalbooleanhasQueuedThreads(){
returnsync.hasQueuedThreads();
}
publicfinalintgetQueueLength(){
returnsync.getQueueLength();
}
protectedCollectiongetQueuedThreads(){
returnsync.getQueuedThreads();
}
publicStringtoString(){
returnsuper.toString()+"[Permits="+sync.getPermits()+"]";
}}
上面对于Semaphore的一些重要内部类和常用方法进行了解释，与CountDownLatch很类似，实现的都是共享的功能，即Semaphore允许得到许可证的线程同时执行，而CountDownLatch允许调用countDown()方法的线程同时执行。并且都是通过内部类实现的。相信看到这里，你能越来越看见AQS为什么被称作JUC包的核心。下面就来介绍一下ReentrantLock
ReentrantLock是如何借助AQS实现锁机制
ReentrantLock是可重入锁，前面博客中写到synchronized实现的锁也是可重入的。不过synchronized是基于JVM指令实现，而ReentrantLock是使用Java代码实现的。ReentrantLock重点就是需要我们手动声明加锁和释放锁，如果手工忘记释放锁，很有可能就会导致死锁，即资源永远都被锁住，其他线程无法得到，当前线程也释放不出去。ReentrantLock实现的是自旋锁，通过循环调用CAS操作实现加锁，避免了线程进入内核态的阻塞状态，所以性能较好。ReentrantLock内部同样实现了公平锁和非公平锁。事实上Synchronized能做的ReentrantLock都能做，但是反过来就不一样了、
经过前面的源码分析我们发现核心的都在当前类的内部类里，而当前类的一些方法不过是使用的内部类以及AQS的方法罢了，所以下面我们就来分析ReentrantLock中的三个内部类。
publicclassReentrantLockimplementsLock,java.io.Serializable{
privatestaticfinallongserialVersionUID=7373984872572414699L;
/**同步的机制都是通过内部类来实现的*/
privatefinalSyncsync;
/**
*在ReentrantLock中state表示的是线程重入锁的次数，当state为0时才能释放锁
*/
abstractstaticclassSyncextendsAbstractQueuedSynchronizer{
privatestaticfinallongserialVersionUID=-5179523762034025860L;
/**
*这个抽象方法提供给公平锁和不公平锁来单独实现，父类不实现
*/
abstractvoidlock();
/**
*首先得到当前线程，而后获取state，如果state为0，也就是没有线程获得当前锁，那么就设置当前线程拥有当前锁的独占访问权，并且返回true。*如果state不为0，那么就看当前线程是否是已经获得过锁的线程，如果是就让state+=acquire，acquire一般是1，即表示线程重入并且返回true。*上面两个条件都不满足就代表是锁被其他线程获取了，当前线程获取不到，所以返回false
*/
finalbooleannonfairTryAcquire(intacquires){
finalThreadcurrent=Thread.currentThread();
intc=getState();
if(c==0){
if(compareAndSetState(0,acquires)){
setExclusiveOwnerThread(current);
returntrue;
}
}
elseif(current==getExclusiveOwnerThread()){
intnextc=c+acquires;
if(nextc<0)//overflow
thrownewError("Maximumlockcountexceeded");
setState(nextc);
returntrue;
}
returnfalse;
}
/**
*先判断当前线程等不等于拥有锁的线程，不等于就会抛异常，也就是释放不了。*等于之后就看state-releases是否为0，当为0的时候就代表释放完全。*可以设置锁的状态为没有线程拥有，从而让锁能被其他线程竞争，否则就设置state，代表线程重入该锁，并且线程还没释放完全。
*/
protectedfinalbooleantryRelease(intreleases){
intc=getState()-releases;
if(Thread.currentThread()!=getExclusiveOwnerThread())
thrownewIllegalMonitorStateException();
booleanfree=false;
if(c==0){
free=true;
setExclusiveOwnerThread(null);
}
setState(c);
returnfree;
}
/*
*该方法检验当前线程是否是锁的独占者
*/
protectedfinalbooleanisHeldExclusively(){
returngetExclusiveOwnerThread()==Thread.currentThread();
}
/*
*该方法是创建一个条件锁，本文不做具体分析
*/
finalConditionObjectnewCondition(){
returnnewConditionObject();
}
//Methodsrelayedfromouterclass
finalThreadgetOwner(){
returngetState()==0?null:getExclusiveOwnerThread();
}
finalintgetHoldCount(){
returnisHeldExclusively()?getState():0;
}
finalbooleanisLocked(){
returngetState()!=0;
}
/**
*使得该类从流中能重构实例，并且会重置为解锁状态
*/
privatevoidreadObject(java.io.ObjectInputStreams)
throwsjava.io.IOException,ClassNotFoundException{
s.defaultReadObject();
setState(0);
}
}
/**
*Sync的不公平版本
*/
staticfinalclassNonfairSyncextendsSync{
privatestaticfinallongserialVersionUID=7316153563782823691L;
/**
*将state从0更新到1成功的话就让当前线程获取锁，否则就会尝试获得锁和获取当前节点的前一节点，并判断这一个节点是否为头节点，即当前线程是不是头节点的直接后继。*如果两个中有一个失败则线程中断，进入阻塞状态
*/
finalvoidlock(){
if(compareAndSetState(0,1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protectedfinalbooleantryAcquire(intacquires){
returnnonfairTryAcquire(acquires);
}
}
/**
*Sync的公平版本
*/
staticfinalclassFairSyncextendsSync{
privatestaticfinallongserialVersionUID=-3000897897090466540L;
/**
*尝试获得锁和获取当前节点的前一节点，并判断这一个节点是否为头节点，即当前线程是不是头节点的直接后继，如果两个中有一个失败则线程中断，进入阻塞状态。*也就是一定按照队列中线程的顺序来实现
*/
finalvoidlock(){
acquire(1);
}
/**
*跟不公平的版本相比其实是在state为0的时候检查当前线程是不是在队列的头部节点的直接后继，来达到公平的概念
*/
protectedfinalbooleantryAcquire(intacquires){
finalThreadcurrent=Thread.currentThread();
intc=getState();
if(c==0){
if(!hasQueuedPredecessors()&&
compareAndSetState(0,acquires)){
setExclusiveOwnerThread(current);
returntrue;
}
}
elseif(current==getExclusiveOwnerThread()){
intnextc=c+acquires;
if(nextc<0)
thrownewError("Maximumlockcountexceeded");
setState(nextc);
returntrue;
}
returnfalse;
}
}
}
ReentrantLock和上面两个类最不同的莫过于ReentrantLock使用的是独占功能，即每次只能有一个线程来获取ReentrantLock类。ReentrantLock类下还有很多方法，这里就不一一介绍，但是本质都是内部类中的实现以及AQS的一些调用
总结：
AQS只是一个基础的框架，里面最核心的就是维护了state变量和CHL队列，而其他的类全部都是通过继承的方法进行扩展，虽然没有直接说源码，但是通过上面三个主要类的源码分析再去看AQS已经不是难事。继承主要改变的就是获取和释放的方法，通过这两个方法来对state和队列进行操作达到我们能够进行的并发控制的功能，事实上J.U.C包下的类和能够实现的功能远不止这三个，后面会选择重点的来介绍。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。