概念
AbstractQueuedSynchronizer抽象类(以下简称AQS)是整个java.util.concurrent
包的核心。在JDK1.5时,Doug Lea引入了J.U.C包,该包中的大多数同步器都是基于AQS来构建的。AQS框架提供了一套通用的机制来管理同步状态(synchronization state)、阻塞/唤醒线程、管理等待队列。如常用的 ReentrantLock/Semaphore/CountDownLatch都是依赖于它。
AQS框架,分离了构建同步器时的一系列关注点,它的所有操作都围绕着资源——同步状态(synchronization state)来展开,并替用户解决了如下问题:
- 资源是可以被同时访问?还是在同一时间只能被一个线程访问?(共享/独占功能)
- 访问资源的线程如何进行并发管理?(等待队列)
- 如果线程等不及资源了,如何从等待队列退出?(超时/中断)
AQS框架将剩下的一个问题留给用户:
什么是资源?如何定义资源是否可以被访问?
这其实是一种典型的模板方法设计模式:父类(AQS框架)定义好骨架和内部操作细节,具体规则(比如资源的获取/释放方式)由子类去实现。
资源访问模式
支持独占模式和共享模式
Exclusive 独占资源
Exclusive(独占,只有一个线程能执行,如 ReentrantLock)
Share 共享资源
Share(共享,多个线程可同时执行,如 Semaphore/CountDownLatch)。
框架结构
原理
如果被请求的共享资源空闲,则将当前请求资源的线程设置为有效的工作线程,并将共享资源设置为锁定状态,如果被请求的共享资源被占用,那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制,这个机制AQS是用CLH队列锁实现的,即将暂时获取不到锁的线程加入到队列中。AQS是将每一条请求共享资源的线程封装成一个CLH锁队列的一个结点(Node),来实现锁的分配。
CLH(Craig,Landin,and Hagersten)队列,是单向链表队列。AQS中的队列是CLH变体的虚拟双向队列(FIFO),虚拟的双向队列即不存在队列实例,仅存在节点之间的关联关系。
AQS使用一个Volatile的int类型的成员变量来表示同步状态,通过内置的FIFO队列来完成资源获取的排队工作,通过CAS完成对State值的修改。
同步状态
同步状态,其实就是资源。AQS使用单个int(32位)来保存同步状态,并暴露出getState、setState以及compareAndSetState操作来读取和更新这个状态。
// java.util.concurrent.locks.AbstractQueuedSynchronizer
private volatile int state;
protected final int getState() {
return state;
}
protected final void setState(int newState) {
state = newState;
}
/**
* 以原子的方式更新同步状态.
* 利用Unsafe类实现
*/
protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
这几个方法都是Final修饰的,说明子类中无法重写它们。可以通过修改State字段表示的同步状态来实现多线程的独占模式和共享模式(加锁过程)。对于自定义的同步工具,需要自定义获取同步状态和释放状态的方式,也就是AQS架构图中的第一层:API层
线程的阻塞/唤醒
在JDK1.5之前,除了内置的监视器机制外,没有其它方法可以安全且便捷得阻塞和唤醒当前线程。JDK1.5以后,java.util.concurrent.locks包提供了LockSupport类来作为线程阻塞和唤醒的工具。
等待队列
等待队列,是AQS框架的核心,整个框架的关键其实就是如何在并发状态下管理被阻塞的线程。
等待队列是严格的FIFO队列,是Craig,Landin和Hagersten锁(CLH锁)的一种变种,采用双向链表实现,因此也叫CLH队列。
结点定义
CLH队列中的结点是对线程的包装,结点一共有两种类型:独占(EXCLUSIVE)和共享(SHARED)。
static final class Node {
/** 共享 */
static final Node SHARED = new Node();
/** 独占 */
static final Node EXCLUSIVE = null;
}
每种类型的结点都有一些状态,其中独占结点使用其中的CANCELLED(1)、SIGNAL(-1)、CONDITION(-2),共享结点使用其中的CANCELLED(1)、SIGNAL(-1)、PROPAGATE(-3)。
AQS使用CLH队列实现线程的结构管理,而CLH结构正是用前一结点某一属性表示当前结点的状态,之所以这种做是因为在双向链表的结构下,这样更容易实现取消和超时功能。
static final class Node {
/** 取消。表示后驱结点被中断或超时,需要移出队列 */
static final int CANCELLED = 1;
/** 发信号。表示后驱结点被阻塞了(当前结点在入队后、阻塞前,应确保将其prev结点类型改为SIGNAL,以便prev结点取消或释放时将当前结点唤醒。) */
static final int SIGNAL = -1;
/** Condition专用。表示当前结点在Condition队列中,因为等待某个条件而被阻塞了 */
static final int CONDITION = -2;
/**
* 传播。适用于共享模式(比如连续的读操作结点可以依次进入临界区,设为PROPAGATE有助于实现这种迭代操作。)
*/
static final int PROPAGATE = -3;
/**
* INITAL: 0 - 默认,新结点会处于这种状态。
* CANCELLED: 1 - 取消,表示后续结点被中断或超时,需要移出队列;
* SIGNAL: -1- 发信号,表示后续结点被阻塞了;(当前结点在入队后、阻塞前,应确保将其prev结点类型改为SIGNAL,以便prev结点取消或释放时将当前结点唤醒。)
* CONDITION: -2- Condition专用,表示当前结点在Condition队列中,因为等待某个条件而被阻塞了;
* PROPAGATE: -3- 传播,适用于共享模式。(比如连续的读操作结点可以依次进入临界区,设为PROPAGATE有助于实现这种迭代操作。)
*
* waitStatus表示的是后续结点状态,这是因为AQS中使用CLH队列实现线程的结构管理,而CLH结构正是用前一结点某一属性表示当前结点的状态,这样更容易实现取消和超时功能。
*/
volatile int waitStatus;
/**
* 前驱指针,用于在结点(线程)被取消时,让当前结点的前驱直接指向当前结点的后驱完成出队动作。
*/
volatile Node prev;
/**
* 后驱指针,用于维护队列顺序,当临界区的资源被释放时,头结点通过next指针找到队首结点。
*/
volatile Node next;
}
队列定义
对于CLH队列,当线程请求资源时,如果请求不到,会将线程包装成结点,将其挂载在队列尾部:
- 初始状态,队列head和tail都指向空
- 首个线程入队,先创建一个空的头结点,然后以自旋的方式不断尝试插入一个包含当前线程的新结点
/**
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}