Java同步器AbstractQueuedSynchronizer--AQS_JAVA_编程开发_程序员俱乐部

?

class="MsoNormal" style="line-height: 25.2px;">Java同步器AbstractQueuedSynchronizer简称AQS（文中全称和简写混用），在java.util.concurrent包中很多依赖状态的API都是基于AQS实现的，比如常用的：ReentrantLock、Semaphore、CountDownLatch、ThreadPoolExecutor等等。

?

可以说AQS是java并发包实现的基石，深入理解AQS可以帮助我们更好的是理解java并发api，而不仅仅停留在使用上。同时我们也可以基于AQS实现一些自定义的可阻塞类，虽然大部分时候不需要我们这样做，因为使用现有的api基本已经足够了。

?

AbstractQueuedSynchronizer直译过来是“抽象的同步队列”，也就是说AQS本质上是维护一个队列，至于你想用这个队列来做什么AQS不管，具体由子类来定。既然AQS是抽象类，就必须要有子类去实现，但在java.util.concurrent包中没有直接直接实现AQS的子类api，其子类都是作为私有的内部类，在各个API使用。下图为jdk1.8中对AQS实现的子类（在jdk1.6中更多）：

?

可以看到一共有11个子类，并且都内部类。AQS在java.util.concurrent并发包中，主要有以下4个功能：

1、实现锁：ReentrantLock（重入锁）、ReentrantReadWriteLock（重入读写锁），并同时提供公平锁和非公平锁实现。

2、实现信号量：Semaphore（主要用于控制某个资源，可以被同时访问的线程个数），并同时支持公平和非公平实现。

3、实现闭锁：CountDownLatch，一种同步辅助工具，可以实现：在一组其他线程中执行的操作完成之前，阻塞一个或多个线程；并在完成之后，同时唤起被阻塞的一个或者多个线程。

4、实现线程池执行器：ThreadPoolExecutor，这时java线程池框架的基石。

?

AQS的基本构成

?

节点定义类Node

前面提到AQS本质上是维护一个队列，首先来看下这个队列中的成员：Node，它是AQS定义的内部类，主要成员变量信息如下：

Java代码??

1. static?final?class?Node?{??
2. ???
3. ????volatile?int?waitStatus;//节点状态??
4. ???
5. ????volatile?Node?prev;//前指针?指向前一个节点??
6. ???
7. ????volatile?Node?next;//后指针指向后一个节点??
8. ???
9. ????volatile?Thread?thread;//线程实例??
10. ???
11. Node?nextWaiter;//下一个等待节点??
12. ???
13. /**状态列表，对应waitStatus字段值*/??
14. ????//共享类型??
15. ????static?final?Node?SHARED?=?new?Node();??
16. ????//独占类型??
17. ????static?final?Node?EXCLUSIVE?=?null;??
18. ????//线程取消类型??
19. ????static?final?int?CANCELLED?=??1;??
20. ????//线程唤醒状态类型?对应condition的?signal、signalAll方法??
21. ????static?final?int?SIGNAL????=?-1;??
22. ????//线程阻塞状态类型，对应condition的await方法??
23. ????static?final?int?CONDITION?=?-2;??
24. ????//对应共享类型释放资源时，传播唤醒线程状态??
25. ????static?final?int?PROPAGATE?=?-3;??
26. ?????
27. ?????
28. ????//省略其他??
29. }??
30. ???

?

可以看到每个节点中都有一个“线程实例”，以及该线程所处的状态waitStatus，以及用于表示前后指针的成员prev、nex，即AQS是双向链表，节点里的“线程实例”其实就是阻塞的线程列表。AQS的主要方法其实就是操作和维护这个双向链表。

?

成员变量

在来看下AQS的主要成员变量：

? ?

Java代码??

1. //队列的头节点??
2. ???private?transient?volatile?Node?head;??
3. ??
4. ???//队列的尾节点??
5. ???private?transient?volatile?Node?tail;??
6. ??
7. ???//资源?状态??
8. ???private?volatile?int?state;??

?

其中最重要的就是队列状态state（注意跟Node中的节点状态区分开），这个字段一般由于表示一种共享的“资源”的状态，比如：在ReentrantLock锁的实现中，它表示锁是否被占用（为0是表示可用）；在信号量Semaphore的实现中，表示剩余的“许可数量”，大于0表示可用；CountDownLatch中表示需全部完成的工作数量。

?

AQS的核心功能就是：在多线程竞争有限的“资源”的情况下，只允许部分线程（或单个线程）访问这种“资源”，并阻塞其他“线程”。AQS的所有方法几乎都是在根据“资源”的状态，操作和维护一个“双向链表”。

?

主要方法

前面提到过，由于资源是有限的，所以AQS的核心方法分为两大类：获取“资源”方法和释放“资源”方法，同时这两类方法又都有公平和非公平实现。另外对应 获取“资源”方法，还有延时获取和可中断获取。由于篇幅有限这里不贴出所有的方法，只分类讲解入口方法的主要流程，如果要深入每个方法，可以通过这些入口方法断点跟下去即可。下面分别来看：

?

独占获取和释放方法

独占获取方法acquire，对应的释放方法release。独占的方式获取“资源”，那这个“资源”状态state其实就只有两种：可以、非可用，一般用于实现独占锁。首先来看acquire方法：

Java代码??

1. public?final?void?acquire(int?arg)?{??
2. ????????if?(!tryAcquire(arg)?&&??
3. ????????????????acquireQueued(addWaiter(Node.EXCLUSIVE),?arg))??
4. ????????????selfInterrupt();??
5. }??
6. ???

?

简易流程如下：

?

通过调用tryAcquire(arg)方法尝试获取“资源”，如果获取到 该线程继续执行；否则调用acquireQueued方法加入队列，并阻塞该线程，注意节点是Node.EXCLUSIVE独占方式。其中tryAcquire(arg)方法是交给子类去实现的，在jdk的api中独占方式一般都是用于获取“锁”，具体的实现可以看下ReentrantLock、ReentrantReadWriteLock中的写锁、ThreadPoolExecutor。

?

再来看下释放资源方法release:

Java代码??

1. public?final?boolean?release(int?arg)?{??
2. ????????if?(tryRelease(arg))?{??
3. ????????????Node?h?=?head;??
4. ????????????if?(h?!=?null?&&?h.waitStatus?!=?0)??
5. ????????????????unparkSuccessor(h);??
6. ????????????return?true;??
7. ????????}??
8. ????????return?false;??
9. }??
10. ???

?

简易流程如下：

?
?

通过调用tryRelease(arg)方法尝试释放资源，其实就是修改“资源”状态state；如果资源释放成功就调用unparkSuccessor方法，唤醒队列头结点的线程（具体是在上述acquireQueued方法中进行阻塞的），继续执行。同样?tryRelease(arg)方法是在子类中实现。

?

共享获取和释放方法

共享获取方法acquireShared，对应的释放方法为releaseShared。在java的api中主要用于实现：ReentrantReadWriteLock中的读锁（共享锁）、Semaphore。首先看下共享获取方法acquireShared：

Java代码??

1. public?final?void?acquireShared(int?arg)?{??
2. ????????if?(tryAcquireShared(arg)?<?0)??
3. ????????????doAcquireShared(arg);??
4. }??

?

通过tryAcquireShared(arg)方法获取“资源”，如果获取到“资源”直接返回，该线程继续执行。如果没有获取到资源，就通过调用doAcquireShared加入队列，并阻塞该线程。这里tryAcquireShared(arg)方法是有子类实现的，具体实现可以参考ReentrantReadWriteLock中的读锁、Semaphore。ReentrantReadWriteLock的获取读锁实现要稍微复杂些，需要判断是否被读锁占有；Semaphore的实现比较简单，就是判断剩余“许可数量”是否大于0。再来看下释放方法releaseShared：

Java代码??

1. public?final?boolean?releaseShared(int?arg)?{??
2. ????????if?(tryReleaseShared(arg))?{??
3. ????????????doReleaseShared();??
4. ????????????return?true;??
5. ????????}??
6. ????????return?false;??
7. }??

?

通过tryReleaseShared(arg)方法尝试释放资源，如果释放成功，调用doReleaseShared方法唤醒在上述acquireShared?方法中阻塞的线程。同样的tryReleaseShared(arg)方法是在子类中实现的。

?

共享获取、排它独占获取，二者几乎没有区别，只是在释放方法中有区别，独占方式释放资源后，只会唤醒“头结点”的线程；而共享方式，会遍历队列，把满足条件的线程全部唤醒，具体可以参考doReleaseShared和acquireQueued方法。

?

可中断获取方法

可中断获取方法acquireInterruptibly（独占可中断获取）、acquireSharedInterruptibly(共享可中断获取)，这两个方法中的独占和共享的实现与上述讲述过程相同，不再累述。如果获取不成功就进入排队并阻塞当前线程，唯一的区别就是会抛出InterruptedException异常，也就是说可以在其他线程调用该线程的interrupt方法，中断该线程放弃任务执行，从而放弃排队。以acquireInterruptibly方法实现为例，如下：

Java代码??

1. public?final?void?acquireInterruptibly(int?arg)??
2. ????????????throws?InterruptedException?{??
3. ????????if?(Thread.interrupted())??
4. ????????????throw?new?InterruptedException();??
5. ????????if?(!tryAcquire(arg))??
6. ????????????doAcquireInterruptibly(arg);??
7. }??

?

与独占获取方法一样 通过tryAcquire(arg)方法获取资源，如果没有获取到就调用doAcquireInterruptibly(arg)方法把当前线程加入队列，并阻塞当前线程。

?

延迟获取方法

延迟获取方法tryAcquireNanos(独占延迟获取)、tryAcquireSharedNanos(共享延迟获取)，这两个方法中的独占和共享的实现与上述讲述过程相同，不再累述。如果获取不成功就进入排队并阻塞当前线程，唯一的区别就是这里阻塞会有时间限制，如果超时就抛出InterruptedException异常，中断该线程放弃任务执行，从而放弃排队。

?

注意与“可中断获取方法”的区别：前者是时间到了，自动中断线程放弃排队；后者是需要外部手动触发。这里以tryAcquireNanos(独占延迟获取)为例，代码实现如下：

Java代码??

1. public?final?boolean?tryAcquireNanos(int?arg,?long?nanosTimeout)??
2. ????????????throws?InterruptedException?{??
3. ????????if?(Thread.interrupted())??
4. ????????????throw?new?InterruptedException();??
5. ????????return?tryAcquire(arg)?||??
6. ????????????doAcquireNanos(arg,?nanosTimeout);??
7. }??

?

与独占获取方法一样 通过tryAcquire(arg)方法获取资源，如果没有获取到就调用doAcquireNanos方法把该线程加入队列，并阻塞该线程，与acquireQueued方法不同地方就是阻塞会有一个时间限制，当阻塞时间到达时抛出InterruptedException异常。

?

可以看到AQS的核心方法分为三类：入口方法，tryXXX方法，doXXX方法。

入口方法：前面已经列出；

tryXXX方法种：有4个方法是交给子类实现的：tryAcquire（尝试独占获取资源）、tryRelease（尝试独占释放资源）、tryAcquireShared（尝试共享获取资源）、tryReleaseShared（尝试释放共享资源）。

doXXX方法：是具体的实现加入队列，以及阻塞线程的核心实现，由于这些方法的代码比较长，就不贴出来了，可以根据上述思路自行查阅jdk API源码。

?

公平和非公平

?

所谓公平，就是所有的线程来获取“资源”时，都得按照FIFO的原则排队；所谓非公平，也不是说不排队，而是先检查“资源”是否可用，如果可用就插队立即使用，否则再进行排队。另外公平和非公平实现，都是在子类中实现的，在AQS中没有实现。比如ReentrantLock公平锁和非公平锁实现。

?

内部类?ConditionObject

最后提下AQS中的内部类：“条件队列”实现类ConditionObject，这个类是Condition接口的实现类（主要方法：await系列方法、signal、signalAll，对应Object类的wait、notify、notifyAll方法），主要用于结合Lock锁结合使用（通过Lock的newCondition()方法得到）。每一个Lock对应一个Condition条件队列，这个条件队列使用的是AQS中的同一个队列，只是队列Node节点中的waitStatus类型不同。比如ReentrantLock锁，通过lock方法可以向AQS中加入节点，也可以通过Condition的await方法向AQS加入节点，只是节点的类型不同而已。

?

?

由于条件队列Condition和Lock锁密切相关，关于这部分内容后面有时间再单独总结，这里不再继续展开。

?

总结

?

简单的总结AQS，本质上就是维护了一个“双向链表”结构的队列，其中每个节点保存了一个阻塞的线程；其主要作用就是用于多线程竞争有限的资源时，对线程阻塞、排队。另外我们也可以根据AQS实现自己的同步器，当然大部分时候使用现有的API实现类已经足够了。

?

http://moon-walker.iteye.com/blog/2406446