Java 并发编程 ----- AQS（抽象队列同步器）
AQS即AbstractQueuedSynchronizer的缩写，是并发编程中实现同步器的一个框架。框架，框架，重要的事情说三遍，框架就是说它帮你处理了很大一部分的逻辑，其它功能需要你来扩展。想想你使用Spring框架的场景，Spring帮助开发者实现IOC容器的bean依赖管理，标签解析等，我们只需要对bean进行配置即可，其他不用管。
AQS基于一个FIFO双向队列实现，被设计给那些依赖一个代表状态的原子int值的同步器使用。我们都知道，既然叫同步器，那个肯定有个代表同步状态（临界资源）的东西，在AQS中即为一个叫state的int值，该值通过CAS进行原子修改。
在AQS中存在一个FIFO队列，队列中的节点表示被阻塞的线程，队列节点元素有4种类型， 每种类型表示线程被阻塞的原因，这四种类型分别是：
-
CANCELLED: 表示该线程是因为超时或者中断原因而被放到队列中 -
CONDITION: 表示该线程是因为某个条件不满足而被放到队列中，需要等待一个条件，直到条件成立后才会出队 -
SIGNAL: 表示该线程需要被唤醒 -
PROPAGATE： 表示在共享模式下，当前节点执行释放release操作后，当前结点需要传播通知给后面所有节点
由于一个共享资源同一时间只能由一条线程持有，也可以被多个线程持有，因此AQS中存在两种模式，如下：
-
1、独占模式
独占模式表示共享状态值state每次只能由一条线程持有，其他线程如果需要获取，则需要阻塞,如JUC中的
ReentrantLock -
2、共享模式
共享模式表示共享状态值state每次可以由多个线程持有，如JUC中的
CountDownLatch
二、AQS 中的核心数据结构和方法
1、既然AQS是基于一个FIFO队列的框架，那么我们先来看下队列的元素节点Node的数据结构，源码如下：
static final class Node {
/**共享模式*/
static final Node SHARED = new Node();
/**独占模式*/
static final Node EXCLUSIVE = null;
/**标记线程由于中断或超时，需要被取消，即踢出队列*/
static final int CANCELLED = 1;
/**线程需要被唤醒*/
static final int SIGNAL = -1;
/**线程正在等待一个条件*/
static final int CONDITION = -2;
/**
* 传播
*/
static final int PROPAGATE = -3;
// waitStatus只取上面CANCELLED、SIGNAL、CONDITION、PROPAGATE四种取值之一
volatile int waitStatus;
// 表示前驱节点
volatile Node prev;
// 表示后继节点
volatile Node next;
// 队列元素需要关联一个线程对象
volatile Thread thread;
// 表示下一个waitStatus值为CONDITION的节点
Node nextWaiter;
/**
* 是否当前结点是处于共享模式
*/
final boolean isShared() {
return nextWaiter == SHARED;
}
/**
* 返回前一个节点，如果没有前一个节点，则抛出空指针异常
*/
final Node predecessor() throws NullPointerException {
// 获取前一个节点的指针
Node p = prev;
// 如果前一个节点不存在
if (p == null)
throw new NullPointerException();
else
// 否则返回
return p;
}
// 初始化头节点使用
Node() {}
/**
* 当有线程需要入队时，那么就创建一个新节点，然后关联该线程对象，由addWaiter()方法调用
*/
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
/**
* 一个线程需要等待一个条件阻塞了，那么就创建一个新节点，关联线程对象
*/
Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
总结下Node节点数据结构设计，队列中的元素，肯定是为了保存由于某种原因导致无法获取共享资源state而被入队的线程，因此 Node
中使用了 waitStatus
表示节点入队的原因，使用 Thread
对象来表示节点所关联的线程。至于 prev
, next
，则是一般双向队列数据结构必须提供的指针，用于对队列进行相关操作。
2、AQS中的共享状态值
之前提到，AQS是基于一个共享的int类型的state值来实现同步器同步的，其声明如下：
/**
* 同步状态值
*/
private volatile int state;
/**
* 获取同步状态值
*/
protected final int getState() {
return state;
}
/**
* 修改同步状态值
*/
protected final void setState(int newState) {
state = newState;
}
由源码我们可以看出，AQS声明了一个int类型的state值，为了达到多线程同步的功能，必然对该值的修改必须多线程可见，因此，state采用volatile修饰，而且 getState()
和 setState()
方法采用final进行修饰，目的是限制AQS的子类只能调用这两个方法对state的值进行设置和获取，而不能对其进行重写自定义设置/获取逻辑。
AQS中提供对state值修改的方法不仅仅只有 setState()
和 getState()
，还有诸如采用CAS机制进行设置的 compareAndSetState()
方法，同样，该方法也是采用 final
修饰的，不允许子类重写，只能调用。
3、AQS中的tryXXX方法
一般基于AQS实现的同步器，如 ReentrantLock
, CountDownLatch
等，对于state的获取操作，子类只需重写其 tryAcquire()
和 tryAcquireShared()
方法即可，这两个方法分别对应独占模式和共享模式下对state的获取操作；而对于释放操作，子类只需重写 tryRelease()
和 tryReleaseShared()
方法即可。
至于如何维护队列的出队、入队操作，子类不用管，AQS已经帮你做好了。
三、AQS 设计妙处
优秀的项目总会有亮点可挖，AQS也是。小编在看了AQS的源码之后，结合其他作者相关博客，总结了以下两点感觉很优秀的设计点，这是我们应该学习的，前辈总是那么优秀。
1、自旋锁
当我们执行一个有确定结果的操作，同时又需要并发正确执行，通常可以采用自旋锁实现。在AQS中，自旋锁采用 死循环 + CAS
实现。针对AQS中的 enq()
进行讲解：
private Node enq(final Node node) {
// 死循环 + CAS ，解决入队并发问题
/**
* 假设有三个线程同时都需要入队操作，那么使用死循环和CAS可保证并发安全，同一时间只有一个节点安全入队，入队失败的线程则循环重试
*
* 1、如果不要死循环可以吗？只用CAS.
* 不可以，因为如果其他线程修改了tail的值，导致1处代码返回false，那么方法enq方法将推出，导致该入队的节点却没能入队
*
* 2、如果只用死循环，不需要CAS可以吗？
* 不可以，首先不需要使用CAS,那就没必要再使用死循环了，再者，如果不使用CAS，那么当执行1处代码时，将会改变队列的结构
*/
for (;;) {
// 获取尾部节点
Node t = tail;
// 如果还没有初始化，那么就初始化
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
// 刚开始肯定是头指针和尾指针相等
tail = head;
} else {
// 当前结点的前驱节点等于尾部节点
node.prev = t;
// 如果当前尾结点仍然是t,那么执行入队并返回true，否则返回false,然后重试
if (compareAndSetTail(t, node)) { // 1
t.next = node;
return t;
}
}
}
}
首先入队操作要求的最终结果必须是一个节点插入到队列中去，只能成功，不能失败！然而这个入队的操作是需要并发执行的，有可能同时有很多的线程需要执行入队操作，因此我们需要采取相关的线程同步机制。自旋锁采取乐观策略，即使用了CAS中的 compareAndSet()
操作，如果某次执行返回fasle,那么当前操作必须重试，因此，采用for死循环直到成功为止，成功，则break跳出for循环或者直接return操作退出方法。
2、模板方法
在AQS中，模板方法设计模式体现在其 acquire()、release()
方法上，我们先来看下源码：
public final void acquire(int arg) {
// 首先尝试获取共享状态，如果获取成功，则tryAcquire()返回true
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
其中调用 tryAcquire()
方法的默认实现是抛出一个异常，也就是说 tryAcquire()
方法留给子类去实现， acquire()
方法定义了一个模板，一套处理逻辑，相关具体执行方法留给子类去实现。
关于更多模板方法设计模式，可以查阅 谈一谈我对‘模板方法’设计模式的理解（Template）
四、自定义自己的并发同步器
下边以 JDK
文档的一个实例进行介绍：
class Mutex implements Lock, java.io.Serializable {
// 自定义同步器
private static class Sync extends AbstractQueuedSynchronizer {
// 判断是否锁定状态
protected boolean isHeldExclusively() {
return getState() == 1;
}
// 尝试获取资源，立即返回。成功则返回true，否则false。
public boolean tryAcquire(int acquires) {
assert acquires == 1; // 这里限定只能为1个量
if (compareAndSetState(0, 1)) {//state为0才设置为1，不可重入！
setExclusiveOwnerThread(Thread.currentThread());//设置为当前线程独占资源
return true;
}
return false;
}
// 尝试释放资源，立即返回。成功则为true，否则false。
protected boolean tryRelease(int releases) {
assert releases == 1; // 限定为1个量
if (getState() == 0)//既然来释放，那肯定就是已占有状态了。只是为了保险，多层判断！
throw new IllegalMonitorStateException();
setExclusiveOwnerThread(null);
setState(0);//释放资源，放弃占有状态
return true;
}
}
// 真正同步类的实现都依赖继承于AQS的自定义同步器！
private final Sync sync = new Sync();
//lock<-->acquire。两者语义一样：获取资源，即便等待，直到成功才返回。
public void lock() {
sync.acquire(1);
}
//tryLock<-->tryAcquire。两者语义一样：尝试获取资源，要求立即返回。成功则为true，失败则为false。
public boolean tryLock() {
return sync.tryAcquire(1);
}
//unlock<-->release。两者语文一样：释放资源。
public void unlock() {
sync.release(1);
}
//锁是否占有状态
public boolean isLocked() {
return sync.isHeldExclusively();
}
}
实现自己的同步类一般都会自定义同步器（sync），并且将该类定义为内部类，供自己使用；而同步类自己（Mutex）则实现某个接口，对外服务。当然，接口的实现要直接依赖sync，它们在语义上也存在某种对应关系！！而sync只用实现资源state的获取-释放方式tryAcquire-tryRelelase，至于线程的排队、等待、唤醒等，上层的AQS都已经实现好了，我们不用关心。
除了Mutex，ReentrantLock/CountDownLatch/Semphore这些同步类的实现方式都差不多，不同的地方就在获取-释放资源的方式tryAcquire-tryRelelase。掌握了这点，AQS的核心便被攻破了！
正文到此结束
热门推荐
相关文章
Loading...
![[HBLOG]公众号](http://www.liuhaihua.cn/img/qrcode_gzh.jpg)
