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的介绍和原理分析