AbstractQueuedSynchronizer 抽象队列同步器 ,简称为 AQS ,可用于构建 阻塞锁 或者其他相关 同步器 的基础框,是Java并发包的基础工具类。通过 AQS 这个框架可以对 同步状态原子性管理、线程的阻塞和解除阻塞、队列的管理 进行统一管理。
AQS 是抽象类,并不能直接实例化,当需要使用 AQS 的时候需要继承 AQS 抽象类并且重写指定的方法,这些重写方法包括 线程获取资源和释放资源的方式 (如ReentractLock通过分别重写线程获取和释放资源的方式实现了公平锁和非公平锁), 同时子类还需要负责共享变量state的维护,如当state为0时表示该锁没有被占,大于0时候代表该锁被一个或多个线程占领(重入锁) ,而队列的维护(获取资源失败入队、线程唤醒、线程的状态等)不需要我们考虑, AQS 已经帮我们实现好了。 AQS 的这种设计模式采用的正是 模板方法模式 。
总结起来子类的任务有:
CAS 操作维护共享变量 state 。 如果对CAS和Java内存模型还不清楚的,建议先了解这两者之后再食用本文,效果更佳! CAS原理分析及ABA问题详解 什么是Java内存模型?
完成以上三个任务即可实现自己的锁。
AQS 作为 J.U.C 的工具类,面向的是需要实现 锁的实现者 ,而锁面向的是 锁的使用者 ,这两者的区别还是需要搞清楚的。
先看 AQS 有哪些重要的成员变量。
// 头结点,你直接把它当做 当前持有锁的线程 可能是最好理解的
private transient volatile Node head;
// 阻塞的尾节点,每个新的节点进来,都插入到最后,也就形成了一个链表
private transient volatile Node tail;
// 这个是最重要的,不过也是最简单的,代表当前锁的状态,0代表没有被占用,大于0代表有线程持有当前锁
// 之所以说大于0,而不是等于1,是因为锁可以重入嘛,每次重入都加上1
private volatile int state;
// 代表当前持有独占锁的线程,举个最重要的使用例子,因为锁可以重入
// reentrantLock.lock()可以嵌套调用多次,所以每次用这个来判断当前线程是否已经拥有了锁
// if (currentThread == getExclusiveOwnerThread()) {state++}
private transient Thread exclusiveOwnerThread; //继承自AbstractOwnableSynchronizer
然后再看看 AQS 的内部结构, AQS 内部数据结构为一个 双向链表 和一个 单向链表 ,双链表为同步队列,队列中的每个节点对应一个 Node 内部类, AQS 通过控制链表的节点而达到阻塞、同步的目的,单链表为普通队列, 可以把同步队列理解为储存阻塞状态的线程,而普通队列储存的是等待状态的线程 ,普通队列中的线程并不能直接去获取资源,而要先从普通队列转到同步队列中排队获取,一个线程要么是在同步队列中,要么是在普通队列中,不可能同时存在这两个队列里面。
Java 阻塞状态 和 等待状态 的线程从Linux内核来看,都是阻塞(等待)状态,它们都会让出CPU时间片。Java为了方便管理线程将“阻塞(等待)”状态细分成了阻塞状态和等待状态,这两个状态的区别 在于由谁去唤醒 ,是操作系统还是其他线程。Java线程请求某一个资源失败的时候就会进入 阻塞状态 ,处于阻塞态的线程会不断请求资源,一旦请求成功,就会进入就绪队列,等待执行。而当线程调用 wait 、 join 、 pack 函数时候会进入 等待状态 ,需要其它线程显性的唤醒否则会无限期的处于等待状态。
Java线程6状态图:
内部类 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;
//代表节点(线程)在 Condition queue中,等待某一条件
static final int CONDITION = -2;
//代表当前节点的后继节点(线程)会传传播唤醒的操作,仅在共享模式下才有作用
static final int PROPAGATE = -3;
//代表当前节点的状态,它的取值除了以上说的CANCELLED、SIGNAL、CONDITION、PROPAGATE,同时
//还可能为0,为0的时候代表当前节点在sync队列中,阻塞着排队获取锁。
volatile int waitStatus;
//当前节点的前驱节点
volatile Node prev;
//当前节点的后继节点
volatile Node next;
//当前节点关联的线程
volatile Thread thread;
//在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;
}
}
每个线程都关联一个节点,节点的状态也代表着线程的状态, AQS 通过对同步队列的管理而达到对线程的管理。
AQS 提供了 2 大功能,基于双链表的同步队列和基于单链表的条件队列,同步队列维护的是 阻塞状态的线程对应的节点 ,这些线程都是阻塞着排队获取锁的,普通队列维护的是 等待状态的线程对应的节点 。
AQS 提供了两种方式去获取资源,分别是 共享模式 和 独占模式 ,但是一般锁只会去继承其中一种模式,不会在一个锁里同时存在 共享模式 和 独占模式 两种模式。
资源指锁、IO、Socket等
当一个线程以共享模式或独占模式去获取资源的时候,如果获取失败则将该线程封装成 Node 节点(同时将该节点标识为共享模式或独占模式) 加入到同步队列的尾部 , AQS 实时维护着这个同步队列,这个队列以 FIFO(先进先出)来管理节点的排队 ,即资源的转移(获取再释放)的顺序是从头结点开始到尾节点。
共享模式和独占模式去获取、释放资源都分别对应着一套 API ,以下分别分析这两套 API
独占模式即获取资源的 排他锁 ,共享模式及获取资源的 共享锁 。
独占模式即一个线程获取到资源后,其他线程不能再对资源进行任何操作,只能阻塞获得资源。
tryAcquire 方法获取资源,如果获取成功,则流程结束,否则继续往下执行。 addWaiter 方法(详细过程看下面的源码解析),将该线 程封装成Node节点 ,并添加到队列 队尾 。 acquireQueued 方法让节点以”死循环”方式进行获取资源,为什么死循环加了双引号呢?因为循环并不是一直让节点无间断的去获取资源,节点会经历 获取资源->失败->线程进入等待状态->唤醒->获取资源……, 线程在死循环的过程会不断等待和唤醒 ,节点进入到自旋状态(详细过程看下面的源码解析), 再循环过程中还会将标识为取消的前驱节点移除队列,同时标识前驱节点状态为SIGNAL 。 LockSupport.lock() 方法实现的,这个方法会相应 Thread.interrupt ,但是不会抛出InterruptedException异常,这点与 Thread.sleep 、 Thread.wait 不一样。
public final void acquire(int arg) {
//该线程调用tryAcquire方法尝试以独占模式获取资源,如果获取失败,则调
//用addWaiter函数,将线程封装到Node节点中,然后再将Node节点加入到同
//步队列的尾部,然后再调用acquireQueued让线程进入到阻塞状态,如果获
//取成功则返回true,然后调用selfInterrupt
//函数。
//注意的是,tryAcquire函数就是继承AQS的子类所需要去重写的方法。
if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE),arg))
selfInterrupt();
}
//AQS的tryAcquire函数并没有获取资源的相关实现,需要继承`AQS`的子类去
//重写这个方法。
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
//创建新的节点,并将线程和节点关联。
//将同步队列的尾节点后继节点指向新节点,
//将新节点的前驱节点指向尾节点,
//新节点称为同步队列的尾节点。
if (pred != null) {
node.prev = pred;
//CAS操作将新节点插入到,成功则返回,不成功则继续下面的enq方法,
//进行死循环CAS插入,直到成功。
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//如果上面的CAS操作插入不成功,则调用enq方法 死循环插入 直到成功。
enq(node);
return node;
}
private Node enq(final Node node) {
//死循环 直到插入成功。
for (;;) {
Node t = tail;
//如果尾节点为null,说明同步队列还未初始化,则CAS操作新建头节点
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
//通过CAS操作将节点插入到同步队列尾部
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
//节点以“死循环”的方式去获取资源,为什么死循环加了双引号呢?因为循环并不
//是一直让节点无间断的去获取资源,节点会经历 获取资源->失败->线程进入等待
//状态->唤醒->获取资源......,线程在死循环的过程会不断等待和唤醒,即节点的自旋。
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//获取节点的前驱节点
final Node p = node.predecessor();
//如果前驱节点为头结点,表示资源正在前驱节点的手上,那该节点
//去尝试获取资源,如果获取成功则将该节点设置为头结点,并且
//返回。
//如果该节点的前驱节点并不是头节点或者是前驱节点是头结点但是
//该节点获取资源失败,继续往下执行。
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//调用shouldParkAfterFailedAcquire函数,将该节点的前驱节点
//的状态设置为SIGNAL,告诉前驱节点我要去“睡觉”了,当资源排
//到你的时候,你就通知我一下让我醒来,即节点做进入等待状态
//的准备。
//当节点做好了进入等待状态的准备,则调用parkAndCheckInterrupt
//函数,让该节点进入到等待状态。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
//获取前驱节点的状态。
int ws = pred.waitStatus;
//如果前驱节点的状态已经为SIGNAL了,即已经做好准备了,那直接返回。
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
//如果前驱节点的状态为取消状态,则将前驱节点移除队列,循环这个过程
//直到前驱节点不为取消状态为止。
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
//如果前驱节点没有做好准备(标志状态为SIGNAL)、前驱节点也没有被取消,
//则使用CAS操作将前驱节点的状态更新为SIGNAL,然后返回false,为什么
//是返回false呢?因为CAS操作并不保证一定能更新成功,返回false的目的
//是让acquireQueued函数再执行一次for循环,这个循环第一可以让该节点
//再尝试获取资源(万一成功了呢 是吧),第二是让acquireQueued函数再调用
//一次shouldParkAfterFailedAcquire函数(即本函数)判断节点的前驱节点是
//否已经设置为SIGNAL状态了。
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
//调用LockSupport.park函数将该线程设置为等待状态
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
//注意LockSupport遇到Thread.interrupt是会立刻返回的,但是不会抛出异常InterruptedExcept
//ion,这个需要注意和Thread.wait,Thread.sleep的区别,
//唤醒的时候 会返回该线程是否为中断唤醒的。
return Thread.interrupted();
}
tryRelease 方法进行释放资源,如果释放成功则继续检查线程(节点)的是否有后继节点,有后继几点则去 唤醒 。 unparkSuccessor 方法进行后继节点的唤醒, 如果后继节点为取消状态,则从队列的队尾往前遍历,找到一个离节点最近且不为取消状态的节点进行唤醒,如果后继节点不为取消状态则直接唤醒 。
源码解析 :
public final boolean release(int arg) {
//线程调用tryRelease方法尝试释放资源,如果释放成功则检查该节点是否有后继节点,有的话则
//调用unpacrkSuccessor()方法去唤醒后继节点。
//注意的是,tryRelease函数就是继承AQS的子类所需要去重写的方法。
if (tryRelease(arg)) {
Node h = head;
//头结点(即释放资源的节点)不为空,头结点的状态不为0,代表有后继节点,需要唤醒。
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
//获取头结点状态。
int ws = node.waitStatus;
//如果状态小于0,即代表有后继节点需要唤醒。
if (ws < 0)
//将头结点的状态置为0 因为只需要唤醒一次
compareAndSetWaitStatus(node, ws, 0);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
Node s = node.next;
//如果头结点的后继节点为空 或者 头结点的后继节点处于取消状态,则从尾部开始往前寻找,
//找到一个离头结点最近 且状态不是取消状态的节点。
if (s == null || s.waitStatus > 0) {
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
//如果头结点的后继节点不为取消状态,则直接将后继节点唤醒
if (s != null)
LockSupport.unpark(s.thread);
}
共享模式下,线程无论是 获取资源还是释放资源,都可能会唤醒后继节点 。
tryAcquireShared 方法进行资源获取,获取失败则调用 doAcquireShared 将 线程封装Node节点加入到同步队列队尾 。 doAcquireShared 方法让节点以”死循环”方式进行获取资源,为什么死循环加了双引号呢?因为循环并不是一直让节点无间断的去获取资源,节点会经历获取资源->失败->线程进入等待状态->唤醒->获取资源……,线程在死循环的过程会不断等待和唤醒,节点进入到自旋状态(详细过程看下面的源码解析)。 如果线程节点被唤醒后,且获取资源成功,且后继节点为共享模式,那么会唤醒后继节点……唤醒会一直传递下去,直到后继节点不是共享模式,唤醒的节点同样会去获取资源 ,这点和独占模式不一样。
共享模式资源的获取和独占模式资源的获取流程差不多,就是在获取资源成功后,会唤醒为共享模式的后继节点,然后被唤醒的后继节点也去获取资源 。
public final void acquireShared(int arg) {
//和独占模式的一样,同样是调用子类重写的tryAcquireShared方法以共享模式进行资源获取。
//如果获取失败,则调用doAcquireShared方法将线程封装成Node节点加入到同步队列的队尾,
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}
private void doAcquireShared(int arg) {
//将线程封装到节点中,且将节点加入到队尾中。
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//获取线程(节点)的前驱节点。
final Node p = node.predecessor();
//如果前驱节点为头结点,则该线程尝试获取资源。
if (p == head) {
//获取资源。
int r = tryAcquireShared(arg);
//获取资源成功则将节点设为头结点。
if (r >= 0) {
//获取成功 对后继SHARED节点持续唤醒
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
//和独占模式的一样。
//调用shouldParkAfterFailedAcquire函数,将该节点的前驱节点
//的状态设置为SIGNAL,告诉前驱节点我要去“睡觉”了,当资源排
//到你的时候,你就通知我一下让我醒来,即节点做进入等待状态的准备。
//当节点做好了进入等待状态的准备,则调用parkAndCheckInterrupt
//函数,让该节点进入到等待状态。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
/*
* Try to signal next queued node if:
* Propagation was indicated by caller,
* or was recorded (as h.waitStatus either before
* or after setHead) by a previous operation
* (note: this uses sign-check of waitStatus because
* PROPAGATE status may transition to SIGNAL.)
* and
* The next node is waiting in shared mode,
* or we don't know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
*/
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
//如果节点为共享节点,则调用doReleaseShared函数唤醒后继节点。
if (s == null || s.isShared())
doReleaseShared();
}
}
tryReleaseShared 方法释放资源,释放成功则调用 doReleaseShared 方法进行后继节点的唤醒。
//调用子类重写的tryReleaseShared方法进行以共享模式释放资源,释放失败则调用doReleaseShared。
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
//如果节点标识后继节点需要唤醒,则调用unparkSuccessor方法进行唤醒。
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
普通队列又称等待队列、条件队列等,普通队列的实现是通过 ConditionObject 的内之类来完成的,,一开始就介绍了同步队列普通队列的去,不过这里再啰嗦一下, 可以把同步队列理解为储存阻塞状态的线程,而普通队列储存的是等待状态的线程 ,普通队列中的线程并不能直接去获取资源,而要先从普通队列转到同步队列中排队获取,一个线程要么是在同步队列中,要么是在普通队列中,不可能同时存在这两个队列里面。
/* * 使当前线程进入等待状态,直到以下4种情况任意一个发生: * 1.另一个线程调用该对象的signal(),当前线程恰好是被选中的唤醒线程 * 2.另一个线程调用该对象的signalAll() * 3.另一个线程interrupt当前线程(此时会抛出InterruptedException) * 4.虚假唤醒(源自操作系统,发生概率低) * ConditionObject要求调用时该线程已经拿到了其外部AQS类的排它锁(acquire成功) */ void await() throws InterruptedException; /* * 与await()相同,但是不会被interrupt唤醒 */ void awaitUninterruptibly(); /* * 与await()相同,增加了超时时间,超过超时时间也会停止等待 * 三个方法功能相似,其返回值代表剩余的超时时间,或是否超时 */ long awaitNanos(long nanosTimeout) throws InterruptedException; boolean await(long time, TimeUnit unit) throws InterruptedException; boolean awaitUntil(Date deadline) throws InterruptedException; /* * 唤醒一个正在等待该条件变量对象的线程 * ConditionObject会选择等待时间最长的线程来唤醒 * ConditionObject要求调用时该线程已经拿到了其外部AQS类的排它锁(acquire成功) */ void signal(); /* * 唤醒所有正在等待该条件变量对象的线程 * ConditionObject要求调用时该线程已经拿到了其外部AQS类的排它锁(acquire成功) */ void signalAll();
可以看到,其作用与Object原生的wait()/notify()/notifyAll()很相似,但是增加了更多的功能。下面以awaitUninterruptibly()、signal()为例,阐述一下其内部实现。
线程执行 condition.await() 方法,将节点1从同步队列转移到普通队列中。
线程执行 condition.signal() 方法,将节点1从普通队列中转移到同步队列。
简述AbstractQueuedSynchronizer
一行一行源码分析清楚AbstractQueuedSynchronizer
Java并发包源码学习之AQS框架(四)AbstractQueuedSynchronizer源码分析
【Java并发】详解 AbstractQueuedSynchronizer
AbstractQueuedSynchronizer的介绍和原理分析