JUC锁-核心类AQS源码详解()

  本篇文章为你整理了JUC锁:核心类AQS源码详解()的详细内容,包含有 JUC锁:核心类AQS源码详解,希望能帮助你了解 JUC锁:核心类AQS源码详解。

  同步状态变量:state就是那个共享资源(private volatile int state;) Lock类继承AQS类并定义lock()、unLock()的方法,表示获取锁和释放锁。多线程并发访问同一个lock实例,lock()方法会cas修改state变量,修改成功的线程获得锁,其他线程进入AQS队列等待。

  
没有必要!sync队列是双向链表结构,出队时,head交替方式,只需要修改head和head后继2个节点引用关系;固定head,就要修改head,head后继,以及head后继的后继 共3个节点。显然前者效率更高

  
 

   为什么在cancelAquire的unParkSuccess中,能将node的后继节点unpark() 万一node前面还有signal节点呢?

  
不存在的,因为经过判断得出此时node就是head的后继。并且必须由这个取消节点node来唤醒后继,要不node线程结束后,就没有线程能够唤醒队列里的其他节点了。

  
shouldParkAfterFailedAcquire有一个疑问,如果线程被unPark唤醒后,tryAcquire()失败了,那么线程会再次进入parkAndCheckInterrupt 再次park阻塞起来,那么谁来唤醒线程呢?

  
先说结果:由抢到锁的那个线程来唤醒!
 

  上述的场景是存在的,例如在非公平锁模式中,B线程被A线程唤醒,A结束,B成为head,B去执行tryAcquire(),但此时C线程抢占到锁,B执行tryAcquire()没有拿到锁,再次park阻塞。C线程执行结束后将A唤醒

  
shouldParkAfterFailedAcquire为什么一定要先判断或者修改前置节点状态改为SIGNAL:-1,才会park阻塞?

  
只有将前置节点状态改为SIGNAL,才能确保当前节点可以被前置unPark唤醒。也就是说阻塞自己前先保证一定能够被唤醒。因为代码中:
 

  独占模式下,唤醒后继前先限制:h.waitStatus != 0
 

  共享模式下,唤醒后继前先限制:h.waitStatus=SIGNAL

  
如何理解这里说的中断:acquire()函数以独占模式获取(资源),忽略中断,即线程在aquire过程中,中断此线程是无效的

  
表示本线程在获取资源期间,如果被其他线程中断,本线程不会因为中断而取消获取资源,只是将中断标记传递下去。

  


 When acquired in exclusive mode,

 

   * attempted acquires by other threads cannot succeed. Shared mode

   * acquires by multiple threads may (but need not) succeed. This class

   * does not quot;understand quot; these differences except in the

   * mechanical sense that when a shared mode acquire succeeds, the next

   * waiting thread (if one exists) must also determine whether it can

   * acquire as well. Threads waiting in the different modes share the

   * same FIFO queue.

  

 

  
共享模式:允许多个线程同时获取资源;当一个节点的线程获取共享资源后,需要要通知后继共享节点的线程,也可以获取了。共享节点具有传播性,传播性的目的也是尽快通知其他等待的线程尽快获取锁。

  独占模式:只能够一个线程占有资源,其它尝试获取资源的线程将会进入到队列等待。

  响应中断并终止:线程只要被中断就不会获取资源:两种情况的中断:1、刚尝试获取、2、进入队列中等待,前者立即停止获取,后者执行取消逻辑,等待节点变为取消状态

  
A、B先后进入队列,状态都是0。A获得资源,进入setHeadAndPropagate晋升为head,A进入doReleaseShared尝试唤醒B时,但B还没将A改为signal,因为A还是0,A将状态改为PROPAGATE

  2 AbstractQueuedSynchronizer学习总结

  2.1 AQS要点总结

  对于AbstractQueuedSynchronizer的分析,最核心的就是sync queue的分析。

  每一个节点都是由前一个节点唤醒

  当节点发现前驱节点是head并且尝试获取成功,则会轮到该线程运行。

  condition queue中的节点向sync queue中转移是通过signal操作完成的。

  SIGNAL,表示后面的节点需要运行。

  PROPAGATE:就是为了避免线程无法会唤醒的窘境。因为共享锁会有很多线程获取到锁或者释放锁,所以有些方法是并发执行的,就会产生很多中间状态,而PROPAGATE就是为了让这些中间状态不影响程序的正常运行。

  2.2 细节分析

  2.2.1 插入节点时先更新prev再更新前驱next

  

//addWaiter():

 

  node.prev = pred; // 1 更新node节点的prev域

  if (compareAndSetTail(pred, node)) {

   pred.next = node; //2 更新node前驱的next域

   return node;

  //enq():

  node.prev = t; // 1 更新node节点的prev域

  if (compareAndSetTail(t, node)) {

   t.next = node;//2 更新node前驱的next域

   return t;

  //unparkSuccessor():

  Node s = node.next; //通过.next来直接获取到节点的后继节点,这个节点的后继的prev一定指向节点本身

   //....

   if (s != null)

   LockSupport.unpark(s.thread);

  

 

  
addWaiter() 或者enq()插入节点时,都是先更新节点的prev域,再更新它前驱的next域。那么通过node.next()取到的后继,后继的prev域一定是指向node本身。如果先更新next域,在更新prev域时出现异常,那么通过.next取到不是完整的节点

  unparkSuccessor()唤醒后继时,Node s = node.next; 通过.next来获取node的后继,后继的prev一定指向node本身

  
2.2.2 为什么unparkSuccessor()要从尾部往前遍历

  因为取消节点的next域指向了自身,所以不能从通过next来遍历,但prev是完整的,所以通过prev来遍历。

  2.2.3 AQS的设计,尽快唤醒其他等待线程体现在3个地方

  共享锁的传播性。

  doReleaseShared()中head改变,会循环唤醒head的后继节点。

  线程获取锁失败后入队列并不会立刻阻塞,而是判断是否应该阻塞shouldParkAfterFailedAcquire,如果前继是head,会再给一次机会获取锁。

  3 AQS 简介

  AQS是一个用来构建锁和同步器的框架。理论参考:JUC同步器框架

  三个基本组件相互协作:

  
同步器一般包含两种方法,一种是acquire,另一种是release。acquire操作阻塞调用的线程,直到或除非同步状态允许其继续执行。而release操作则是通过某种方式改变同步状态,使得一或多个被acquire阻塞的线程继续执行。

  3.1 AQS核心思想

  如果请求的共享资源空闲,则将当前请求线程设置为有效工作线程,并且将共享资源设置为锁状态

  设计一套机制:【线程如何阻塞等待以及被唤醒时锁如何分配】?这个机制AQS是用CLH队列锁实现的

  CLH队列锁:一个虚拟的双向队列,AQS是将每条请求共享资源的线程封装成一个CLH锁队列的一个节点(Node)来实现锁的分配。【严格的FIFO队列,框架不支持基于优先级的同步】

  使用一个int成员变量来表示同步状态,使用volatile修饰保证线程可见性,并使用CAS思想进行值维护。

  3.2 对资源的共享方式

  两种方式:

  Exclusive(独占):只有一个线程能执行。又可分为公平锁和非公平锁:

  公平锁:按照线程在队列中的排队顺序,先到者先拿到锁

  非公平锁:当线程要获取锁时,无视队列顺序直接去抢锁,谁抢到就是谁的

  
分析类,首先就要分析底层采用了何种数据结构,抓住核心点进行分析:
 

  Sync queue,即同步队列,是双向链表,包括head节点和tail节点,head节点主要用作后续的调度

  Condition queue不是必须的,其是一个单向链表,只有当使用Condition时,才会存在此单向链表。并且可能会有多个Condition queue

  4 AbstractQueuedSynchronizer源码分析

  4.1 类的继承关系

  

public abstract class AbstractQueuedSynchronizer

 

   extends AbstractOwnableSynchronizer

   implements java.io.Serializable

  

 

  继承自抽象类:AbstractOwnableSynchronizer,父类提供独占线程的设置与获取的方法

  

public abstract class AbstractOwnableSynchronizer

 

   implements java.io.Serializable {

   private static final long serialVersionUID = 3737899427754241961L;

   protected AbstractOwnableSynchronizer() { }// 构造函数

   private transient Thread exclusiveOwnerThread; //独占模式下的线程

   // 设置独占线程

   protected final void setExclusiveOwnerThread(Thread thread) {

   exclusiveOwnerThread = thread;

   // 获取独占线程

   protected final Thread getExclusiveOwnerThread() {

   return exclusiveOwnerThread;

  

 

  4.1.1 AQS需要子类重写的方法

  

 protected boolean tryAcquire(int arg) {//独占方式获取锁

 

   throw new UnsupportedOperationException();

   protected boolean tryRelease(int arg) { //释放独占的锁

   throw new UnsupportedOperationException();

   protected int tryAcquireShared(int arg) { //以共享方式获取锁

   throw new UnsupportedOperationException();

   protected boolean tryReleaseShared(int arg) {//释放共享锁

   throw new UnsupportedOperationException();

   protected boolean isHeldExclusively() {//是否独占资源

   throw new UnsupportedOperationException();

  

 

  关于重写说明:
 

  目的是将共享资源state的读写交给子类管理,AQS专注在队列的维护以及线程的阻塞与唤醒

  4.2 类的常量/成员变量

  

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer

 

   implements java.io.Serializable {

   private static final long serialVersionUID = 7373984972572414691L;

   // 头节点

   private transient volatile Node head;

   // 尾节点

   private transient volatile Node tail;

   //0:表示没有线程获取到锁;1表示有线程获取到锁;大于1:表示有线程获得了锁,且允许重入

   private volatile int state;

   // 自旋时间

   static final long spinForTimeoutThreshold = 1000L;

   // 以下跟cas有关

   private static final Unsafe unsafe = Unsafe.getUnsafe(); // Unsafe类实例

   private static final long stateOffset; // state内存偏移地址

   private static final long headOffset; // head内存偏移地址

   private static final long tailOffset; // state内存偏移地址

   private static final long waitStatusOffset;// tail内存偏移地址

   private static final long nextOffset; // next内存偏移地址

   // 静态初始化块

   static {

   try {

   stateOffset = unsafe.objectFieldOffset

   (AbstractQueuedSynchronizer.class.getDeclaredField("state"));

   headOffset = unsafe.objectFieldOffset

   (AbstractQueuedSynchronizer.class.getDeclaredField("head"));

   tailOffset = unsafe.objectFieldOffset

   (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));

   waitStatusOffset = unsafe.objectFieldOffset

   (Node.class.getDeclaredField("waitStatus"));

   nextOffset = unsafe.objectFieldOffset

   (Node.class.getDeclaredField("next"));

   } catch (Exception ex) { throw new Error(ex); }

  

 

  说明:

  属性中包含了头节点head,尾节点tail,状态state、自旋时间spinForTimeoutThreshold

  AbstractQueuedSynchronizer抽象的属性在内存中的偏移地址,通过该偏移地址,可以获取和设置该属性的值

  同时还包括一个静态初始化块,用于加载内存偏移地址。

  4.3 静态内部类Node

  线程封装成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;//表示当前节点的后继节点包含的线程需要被运行【被unpark】,

   static final int CONDITION = -2;//表示当前节点在等待condition,也就是在condition队列中

   static final int PROPAGATE = -3;//表示当前场景下后续的acquireShared能够得以执行

   volatile int waitStatus;//节点状态;表示当前节点在sync队列中,等待着获取锁

   volatile Node prev; // 指向当前节点的前驱

   volatile Node next;// 指向当前节点的后继

   volatile Thread thread;//节点所对应的线程

   Node nextWaiter;// 下一个等待者 只跟condition有关

   private transient volatile Node head; // 头节点 懒加载

   private transient volatile Node tail; //尾节点 懒加载

   private volatile int state; // 同步状态

   // 节点是否在共享模式下等待

   final boolean isShared() {

   return nextWaiter == SHARED;

   // 获取前驱节点,若前驱节点为空,抛出异常

   final Node predecessor() throws NullPointerException {

   Node p = prev;// 保存前驱节点

   if (p == null)

   throw new NullPointerException();

   else

   return p;

   // 无参构造函数

   Node() { // Used to establish initial head or SHARED marker

   // 构造函数

   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说明:

  每个被阻塞的线程都会被封装成一个Node节点,放入队列。Node包含了一个Thread类型的引用,并且有自己的状态:

  CANCELLED:1,表示当前的线程被取消。

  SIGNAL:-1,表示负责unPark后继【由前一个节点unPark下一个节点】。

  CONDITION:-2,表示当前节点在等待condition,也就是在condition queue中。

  PROPAGATE:-3,表示当前场景下后续的acquireShared能够得以执行。

  默认值:0,发生在:1、节点加入到队列成为tail节点,2、节点成为head,并准备唤醒后继

  4.4 构造函数

  

protected AbstractQueuedSynchronizer() { } //默认的无参构造

 

  

 

  4.5 核心方法分析

  4.5.1 核心方法概览

  

public final void acquireShared(int arg) {...} // 获取共享资源的入口(忽略中断)

 

  protected int tryAcquireShared(int arg); // 尝试获取共享资源

  private void doAcquireShared(int arg) {...} // AQS中获取共享资源流程整合

  private Node addWaiter(Node mode){...} // 将node加入到同步队列的尾部

  protected int tryAcquireShared(int arg); // 尝试获取共享资源

  private void setHeadAndPropagate(Node node, int propagate) {...} // 设置 同步队列的head节点,以及触发"传播"操作

  private void doReleaseShared() {...} // 遍历同步队列,调整节点状态,唤醒待申请节点

  private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {...} // 如果获取资源失败,则整理队中节点状态,并判断是否要将线程挂起

  private final boolean parkAndCheckInterrupt() {...} // 将线程挂起,并在挂起被唤醒后检查是否要中断线程(返回是否中断)

  private void cancelAcquire(Node node) {...} // 取消当前节点获取资源,将其从同步队列中移除

  

 

  4.5.2 acquire()方法

  该函数以独占模式获取(资源),忽略中断。
 

  流程如下:
 

  源码如下:

  

public final void acquire(int arg) {

 

   if (!tryAcquire(arg) acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

   selfInterrupt(); //来到这里,表示线程拿到锁,并且读取到线程的中断标识为true,调用selfInterrupt()来恢复线程的interrupted中断标志(被parkAndCheckInterrupt()擦除了,所以再设置一次)。

  static void selfInterrupt() {

   Thread.currentThread().interrupt();//线程设置interrupted中断标志

  protected boolean tryAcquire(int arg) {

   throw new UnsupportedOperationException();

  

 

  acquire()总结

  先调用tryAcquire(),由子类实现来尝试加锁,如果获取到锁,则线程继续执行;反则,节点加入队列

  调用addWaiter(),将调用线程封装成为一个节点并放入AQS队列。

  调用acquireQueued(),先park阻塞等待,直到被unPark唤醒。

  如果线程被设置中断,那么acquire结束前,需要重新设置中断。

  4.5.3 addWaiter()方法

  addWaiter:快速添加的方式往sync queue尾部添加节点

  

// 添加等待者

 

   private Node addWaiter(Node mode) {

   // 新生成一个节点

   Node node = new Node(Thread.currentThread(), mode);

   // 创建临时引用pred,跟tail指向相同地址

   Node pred = tail;

   if (pred != null) { // 尾节点不为空,即队列已经初始化过

   // 将node的prev域连接到尾节点

   node.prev = pred;

   if (compareAndSetTail(pred, node)) { // cas更新tail,指向新创建的node

   // 设置尾节点的next域为node

   pred.next = node; // 结合 node.prev = pred; 形成双向链表

   return node; // 返回新生成的节点

   enq(node); // 队列还未初始化,或者是compareAndSetTail操作失败,则进入enq

   return node;

   //关于并发情景说明:

   // 从 Node pred = tail; 到 compareAndSetTail(pred, node); 期间,队列可能插入了新的节点,pred指向的不是最新的tail,那么compareAndSetTail(pred, node) 就会执行失败,同时 node.prev = pred; node的前驱也不是最新的tail。

   // 通过enq()来解决并发问题,enq()通过自旋+cas来保证线程安全

  

 

  addWaiter()说明:

  使用快速添加的方式(失败不重试)创建新节点并添加到往队列尾部,更新tail

  如果队列还没有初始化或者cas失败,则调用enq()插入队列

  4.5.4 enq()方法

  

 // 线程安全地创建队列、或者将节点插入队列、

 

   private Node enq(final Node node) {

   for (;;) { // 自旋+cas,确保节点能够成功入队列

   Node t = tail;//尾节点

   if (t == null) { // 尾节点为空,即还没被初始化

   if (compareAndSetHead(new Node())) // 设置head。 !!!!注意,这里是new node,没有使用参数的node,因此head节点不引用任何线程

   tail = head; // 头节点与尾节点都指向同一个新生节点。循环继续,进入else后,参数node将插入到队列

   } else { // 尾节点不为空,即已经被初始化过

   node.prev = t; // 将node节点的prev域连接到尾节点

   if (compareAndSetTail(t, node)) { // 比较更新tail,node成为新的tail

   // 设置尾节点的next域为node

   t.next = node; // 结合 node.prev = t; 形成双向链表

   return t; // 返回Node的前驱节点

   //CAS head field. Used only by enq.

   private final boolean compareAndSetHead(Node update) {

   return unsafe.compareAndSwapObject(this, headOffset, null, update);

   //CAS head field. Used only by enq.

   private final boolean compareAndSetTail(Node expect, Node update) {

   return unsafe.compareAndSwapObject(this, tailOffset, expect, update);

  

 

  enq()方法总结:

  功能:cas+自旋方式将节点插入队列

  如果队列未初始化,先创建头节点head(head不指向任务线程),再将节点插入到队列(当第一个节点被创建后,队列实际有两个节点:head+业务节点)。

  如果队列已经初始化,则直接插入队列

  4.5.5 acquireQueue()方法

  作用:sync队列中的节点在独占且忽略中断的模式下获取(资源)

  源码如下:

  

// sync队列中的节点在独占且忽略中断的模式下获取(资源):

 

   final boolean acquireQueued(final Node node, int arg) {

   // 标志

   boolean failed = true;

   try {

   // 中断标识。如果线程唤醒后,中断标识是true,外层的acquire()将进入selfInterrupt()。

   boolean interrupted = false;

   // 无限循环 :如果前驱不是head,那线程将park阻塞,等待前面的节点依次执行,直到被unPark唤醒

   for (;;) {

   // 获取node的前驱,如果前驱是head,则表明前面已经没有线程等待了,该线程可能成为工作线程

   final Node p = node.predecessor();

   // 前驱为头节点并且成功获得锁

   if (p == head tryAcquire(arg)) {

   setHead(node); // node晋升为head

   p.next = null; // 旧head的next域指向null,将会被GC,移出队列

   failed = false; // 设置标志

   return interrupted; //拿到锁,break循环,并返回中断标识

   //执行到这里,前驱非head 或者 前驱是head但获取锁失败,那么:1、将前驱状态改为signal 2、当前线程unPark阻塞

   //shouldParkAfterFailedAcquire():寻找非取消状态的前驱,如果状态为signal返回true 反则,将前驱状态改为signal、再返回false

   //前驱是signal ,执行parkAndCheckInterrupt()后,当前线程park阻塞。一直到线程被unPark唤醒,再返回线程的中断状态

   if (shouldParkAfterFailedAcquire(p, node)

   parkAndCheckInterrupt())//parkAndCheckInterrupt返回true表明线程中断状态为true

   //上面if同时成立,才会执行。

   interrupted = true; //那么把中断标识置为true

   } finally { //(有异常,在抛出之前执行finally;没有异常,在return之前执行finally)

   if (failed)//只有try的代码块出现异常,failed才会是true。什么情景会产生异常?cancelAcquire分析时有说明

   cancelAcquire(node); //执行取消逻辑

   final Node predecessor() throws NullPointerException {

   Node p = prev;

   if (p == null)

   throw new NullPointerException();

   else

   return p;

   private void setHead(Node node) {

   head = node;

   node.thread = null;//再次表明head的thread属性是空的

   node.prev = null;

  

 

  acquireQueue()总结:

  功能:节点进入AQS队列后,先park阻塞等待,直到被unPark唤醒,或者中断唤醒

  找到非取消状态的前驱(取消状态的将会被移出队列并GC),如果前驱是SIGNAL,那么当前节点进入park阻塞,否则,先将前驱改为SIGNAL,再进入park阻塞。

  被unPark唤醒后,判断前驱是头节点且获取到资源(tryAcquire成功),当前节点晋升为头节点。自此,线程获取到锁

  调用shouldParkAfterFailedAcquire和parkAndCheckInterrupt函数,表明只有当该节点的前驱节点的状态为SIGNAL时,才可以对该节点所封装的线程进行park操作。

  4.5.6 shouldParkAfterFailedAcquire()方法

  

// 当获取(资源)失败后:1、判断能否将当前线程park;2、修改前驱节点状态为signal

 

   private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

   // 获取前驱节点的状态

   int ws = pred.waitStatus;

   if (ws == Node.SIGNAL) // 状态为SIGNAL

   // 只有当前驱节点为 signal时,才返回true ,表示当前线程可以安全地park阻塞;其它情况返回false

   return true;

   //跳过那些CANCELLED状态的前驱

   if (ws 0) { // 表示状态为CANCELLED,为1

   do {

   node.prev = pred = pred.prev;

   } while (pred.waitStatus // 找到pred节点前面最近的一个状态不为CANCELLED的节点;然后跳出循环并返回false

   pred.next = node;

   } else { // 为PROPAGATE -3 或者是0 ,(为CONDITION -2时,表示此节点在condition queue中)

   // cas更新前驱的状态为SIGNAL.如果前驱是头节点,那么头节点ws=SIGNAL

   compareAndSetWaitStatus(pred, ws, Node.SIGNAL);

   // 不能进行park操作

   return false;

   //CAS waitStatus field of a node.

   private static final boolean compareAndSetWaitStatus(Node node,int expect, int update) {

   return unsafe.compareAndSwapInt(node, waitStatusOffset, expect, update);

  

 

  shouldParkAfterFailedAcquire()总结:

  如果前驱状态是:SIGNAL,返回true。表示当前节点可以安全地unPark()阻塞

  遇到取消的前驱节点,则跳过。这些被取消的节点会从队列中移除并GC

  如果前驱状态不是:SIGNAL,将前驱状态改为:SIGNAL,返回false,回到1 继续

  4.5.7 parkAndCheckInterrupt()方法

  

// 进行park操作并且返回该线程的中断标识

 

   private final boolean parkAndCheckInterrupt() {

   LockSupport.park(this); //外面的for循环可能会导致多次park,不过没关系,park允许多次执行

   //被唤醒之后,返回中断标记,即如果是正常唤醒则返回false,如果是由于中断醒来,就返回true

   return Thread.interrupted(); // acquireQueued() 中声明的interrupted 将会被更新为这里的返回结果

   public static boolean interrupted() {

   return currentThread().isInterrupted(true);//返回当前线程interrupted中断标记,同时会清除此interrupted标记

  

 

  方法总结:

  执行park操作(前提:前驱状态是SIGNAL),在队列中阻塞等待。

  被unPark()唤醒后,返回线程的interrupted中断标识,并且清除interrupted标记

  4.5.8 cancelAcquire()方法

  什么时候才会执行cancelAcquire?

  

在lockInterruptibly()会通过抛出中断异常来执行cancelAcquire方法,lock方法过程中则不会执行该代码,作者这么些的意图在于for循环内部如果出现不可控的因素导致产生未知的异常,则会执行cancelAcquire,很明显这属于一种相对偏保守的保险代码。

 

  

 

  

// 取消获取锁

 

   private void cancelAcquire(Node node) {

   // Ignore if node doesnt exist

   if (node == null) // node为空,返回

   return;

   node.thread = null;// thread置空 备注1

   // Skip cancelled predecessors

   Node pred = node.prev;// pred表示:最靠近node并且状态不等于取消的前驱节点

   while (pred.waitStatus 0)

   node.prev = pred = pred.prev; //更新pred,往列头推进

   Node predNext = pred.next; //predNext表示:pred的后继

   // 设置node节点的状态为CANCELLED

   node.waitStatus = Node.CANCELLED; //备注2

   if (node == tail compareAndSetTail(node, pred)) { // 若node节点为尾节点,则pred成为尾节点 备注3

   // pred的next域置为null

   compareAndSetNext(pred, predNext, null);

   } else { // 2、node节点不为尾节点,或者比较设置不成功

   int ws;

   //下面一串判断,最终目标:在node移除队列前,将有效的前驱节点状态改为signal

   if (pred != head

   ((ws = pred.waitStatus) == Node.SIGNAL

   (ws = 0 compareAndSetWaitStatus(pred, ws, Node.SIGNAL)))

   pred.thread != null) {

   // pred节点不为头节点,并且

   //pred节点的状态为SIGNAL)或者

   // pred节点状态小于等于0,并且比较并设置等待状态为SIGNAL成功,并且pred节点所封装的线程不为空

   Node next = node.next;

   if (next != null next.waitStatus = 0) // 后继不为空并且后继的状态小于等于0

   compareAndSetNext(pred, predNext, next); // 比较并设置pred.next = next; 到这里:node的前驱节点指向node的后继节点。 备注4

   } else {

   // 这里,pred==head (3、即node是head的后继)或者pred.status=0,-2时 【前面while (pred.waitStatus 0) 已经限制了pred一定是 =0】,执行:

   unparkSuccessor(node); // 唤醒node的后继

   node.next = node; // help GC 后继节点指向自身 备注5

   //修改参数node的next域

   private static final boolean compareAndSetNext(Node node, Node expect, Node update) {

   return unsafe.compareAndSwapObject(node, nextOffset, expect, update);

  

 

  对cancelAcquire()总结之前,先明确以下两点:

  基于对acquire()方法的分析,调用链:addWait()- enq()- acquireQueue()- cancelAcquire(node),进入到cancelAcquire()时,节点node一定已经在队列中,而且它不会是head,并且没有持有锁。

  AQS通过管理这些属性:waitStatus、thread、prev、next、head、tail、nextWaiter ,成为一个虚拟的列队。

  cancelAcquire(node)总结:

  cancelAcquire()负责将node移出队列,并保持队列中其他节点的顺序关系不变,它做了以下工作:

  waitStatus更新为cancel (备注2)

  thread更新为null(备注1)

  tail:如果node是尾节点,更新tail引用 (备注3)

  head:不需要更新(node不会是head)

  prev:没有更新

  next: node前置的next域更新指向node后继,并且node的next指向了自身 (备注4、备注5)

  nextWaiter:不需要更新(跟condition有关,这里不涉及)

  执行cancelAcquire后,队列变成这样的:
 

  发现:

  node没有移出队列,因为被后继的prev所引用。

  node.next变了,指向了自身,这就能解释为什么unparkSuccessor()是从后往前遍历:因为取消节点的next域指向了自身,所以不能从通过next来遍历,但prev是完整的,所以通过prev来遍历。

  取消节点,暂存在队列中,当后继节点被唤醒,执行shouldParkAfterFailedAcquire后,取消节点的引用链清空,移出队列,最后GC回收。

  4.5.9 unparkSuccessor()方法

  

 // 唤醒node节点的后继

 

   private void unparkSuccessor(Node node) {

   // 获取node节点的等待状态

   int ws = node.waitStatus;

   if (ws 0) // 状态值小于0,为SIGNAL -1 或 CONDITION -2 或 PROPAGATE -3

   // cas节点状态为0

   compareAndSetWaitStatus(node, ws, 0);//如果head没有后继的情况下,状态会一直=0

   Node s = node.next;

   //若后继为空,或后继已取消,则从尾部往前遍历 找到最靠近的一个处于正常阻塞状态的节点进行唤醒

   // 什么时候s==null ? node的后继节点是取消状态时,node.next为null

   if (s == null s.waitStatus 0) {

   s = null;

   // 由尾节点向前倒着遍历队列,但不会超过node节点

   for (Node t = tail; t != null t != node; t = t.prev)

   if (t.waitStatus = 0)

   s = t;

   if (s != null)

   LockSupport.unpark(s.thread);//唤醒s节点线程

  

 

  unparkSuccessor()总结:

  
作用:找到有效的后继节点unPark唤醒

  寻找有效后继时从尾往前倒着遍历:因为取消节点的next域指向了自身,所以不能从通过next来遍历

  将发起unPark唤醒的节点(只能是head)状态改为0(意味着在head唤醒后继,到被后继推出队列的期间,状态变为0)

  
Node h = head;

   // 线程A调用acquire()获取到锁之后,A线程节点变为head,然后A调用release 释放锁,存在两种情况:

   // 1、 如果有新的线程B入队,B成为后继节点,B会将A状态改为SIGNAL,那么(h != null h.waitStatus != 0 )成立,unparkSuccessor()唤醒后继节点

   // 2、如果A后面没有节点,A状态是默认值:0 ,那么h.waitStatus != 0 不成立,直接返回true,不需要唤醒后继节点。

   if (h != null h.waitStatus != 0) // 头节点不为空并且头节点状态不为0

   unparkSuccessor(h); //由head唤醒后继节点

   return true;

   return false;

  

 

 

  release()总结:

  功能:释放独占锁

  先调用tryRelease()由子类实现释放锁

  如果释放锁成功,然后unPark唤醒后继节点(没有后继就不需要唤醒)

  4.5.11 acquireSharedInterruptibly()方法

  

 //获取共享资源,响应中断

 

   public final void acquireSharedInterruptibly(int arg)

   throws InterruptedException {

   if (Thread.interrupted()) //读取线程中断标记,然后擦除标记

   throw new InterruptedException(); //中断标记为true,抛出中断异常,停止执行

   if (tryAcquireShared(arg) 0) //调用子类实现方法 获取资源

   doAcquireSharedInterruptibly(arg); //没有获取到,那么再尝试获取(进入队列排队等待)

  

 

  获取共享资源流程图:
 

  acquireSharedInterruptibly()总结:

  共享模式获取对象,响应中断并终止获取

  先调用子类实现获取资源,没有获取到再加队队列等待。

  4.5.12 doAcquireSharedInterruptibly()方法

  

//获取共享资源,响应中断

 

  private void doAcquireSharedInterruptibly(int arg)

   throws InterruptedException {

   final Node node = addWaiter(Node.SHARED); //增加等待节点

   boolean failed = true;

   try {

   for (;;) {//无限循环,直到r 0

   final Node p = node.predecessor(); // p表示 刚插入节点的前驱

   //1、如果前驱是head

   if (p == head) {

   int r = tryAcquireShared(arg);//调用子类实现方法 尝试获取共享资源

   if (r = 0) { // 0 表示 获取到资源

   // 1、如果是ReentrantReadWriteLock、CountDownLatch ,有可能r=1

   // 2、如果是Semaphore,有可能r=0

   // 1、2 都调用setHeadAndPropagate进行共享传播判断

   setHeadAndPropagate(node, r);// 更新head并进行共享传播

   p.next = null; // 将队列头节点的next域置空,之后,这个节点将被GC回收

   failed = false;

   return;

   // 2、前驱不是head

   //线程park阻塞,直至被unPark唤醒,或者被其它线程中断唤醒

   if (shouldParkAfterFailedAcquire(p, node)

   parkAndCheckInterrupt())

   throw new InterruptedException(); //进入这里表示线程中断标记为true,那么抛出中断异常

   } finally {

   if (failed) //当try 代码块有异常:中断异常 或 其他未知异常,failed才是true

   cancelAcquire(node);//取消获取资源

  

 

  doAcquireSharedInterruptibly()总结:

  创建节点并插入aqs队列,将前驱状态改为signal,park阻塞,等待unPark唤醒。

  正常唤醒后,无限循环直到前驱是head并且调用子类方法获取共享资源成功,调用setHeadAndPropagate()成为head并进行共享传播。

  被中断唤醒、或者循环等待过程发生中断异常,执行cancelAcquire()取消获取资源

  4.5.13 setHeadAndPropagate()方法

  setHeadAndPropagate在获取共享资源的时候被调用

  

// 设置 同步队列的head节点,以及触发"传播"操作

 

  private void setHeadAndPropagate(Node node, int propagate) {

   Node h = head; // 记录更新前的head

   setHead(node); //参数node 成为新的head

   //判断:

   if (propagate 0 h == null h.waitStatus 0

   (h = head) == null h.waitStatus 0) {

   Node s = node.next; //获取node后继

   //后继为空或者后继是等待共享资源的节点

   if (s == null s.isShared())

   doReleaseShared(); //释放共享资源

  

 

  满足调用doReleaseShared的条件分析:

  
propagate 0:
 

  ReentrantReadWriteLock、CountDownLatch 调用tryAcquireShared()返回1进入,满足条件;Semaphore 进入,propagate可能等于0,不满足,继续2

  
h == null:
 

  h == null 表示旧head变为null,程序没有地方设置head=null,并且这里h引用着head意味着head不会被GC。 因此,h == null不满足条件,继续3 【不知道哪种情况下h==null todo】

  
方法功能:设置 同步队列的head节点,以及触发"传播"操作:

  如果head的后继是共享类型节点或者为null,调用doReleaseShared()来唤醒后继

  4.5.14 doReleaseShared()方法

  

//遍历同步队列,调整节点状态,唤醒待申请节点

 

  private void doReleaseShared() {

   for (;;) {

   Node h = head;

   //1、head 不等于 tail 且不等于 null

   if (h != null h != tail) {

   int ws = h.waitStatus;

   if (ws == Node.SIGNAL) { //如果head状态为signal ,cas修改为0

   if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))

   continue; // loop to recheck cases

   unparkSuccessor(h); //唤醒后继

   //如果节点的后继还没有将其前驱改为signal,这里ws==0是成立的

   else if (ws == 0 //如果head状态为0,cas修改为propagate

   !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))

   continue; // 如果在int ws = h.waitStatus; 之后,后继将head节点改为signal,那么cas失败,continue继续循环后, if (ws == Node.SIGNAL) 满足,那么将会唤醒后继。

   // 只有head没有发生变化,循环才会结束,若head改变,继续循环

   if (h == head) // loop if head changed

   break;

  

 

  doReleaseShared()总结:

  如果头节点状态为signal,那么CAS更新头节点状态为0,成功则调用unparkSuccessor()唤醒后继,失败则重试

  如果头节点状态为0,那么将CAS更新头节点状态为PROPAGTATE ,失败则重试。

  最后如果判断head是否发生变化,有变化则重复1、2,没有变化则方法结束。

  PROPAGTATE状态的意义是,增加一个状态判断,当前驱获取资源,后继同时也有机会获取到资源

  4.5.15 releaseShared()方法

  

public final boolean releaseShared(int arg) {

 

   if (tryReleaseShared(arg)) {

   doReleaseShared();

   return true;

   return false;

  

 

  releaseShared()方法总结:

  调用子类的实现方法tryReleaseShared()释放n个共享资源,释放成功则继续调用doReleaseShared()来唤醒队列中的等待节点

  
头节点unPark唤醒后继时,后继节点唤醒后重新进入shouldParkAfterFailedAcquire()

  取消节点后面有新节点入列时,新节点执行shouldParkAfterFailedAcquire()

  以第一个情景为例子分析:
 

  6 在shared模式中为什么需要PROPAGATE状态

  结论:在前驱节点获取资源时,后继也能够有机会申请资源,不需要等待前驱通过releaseShare()来唤醒。
 

  分析如下:

  

1:A B 先后进入队列

 

   2:A被唤醒,获得资源,调用setHeadAndPropagate(),晋升为head

   3、B调用shouldParkAfterFailedAcquire(),尝试将A状态改为signal但未执行

   4、A进入doReleaseShared(),A状态等于0(3还没执行),进入ws == 0 分支处理。

   5、此时3执行完成,B将A的状态改为signal,然后B park阻塞

   6、A执行compareAndSetWaitStatus(h, 0, Node.PROPAGATE)失败,continue继续

   7、A进入(ws == Node.SIGNAL)分支,执行compareAndSetWaitStatus(h, Node.SIGNAL, 0)成功,然后再执行unparkSuccessor(),将B唤醒。

   8、A将B唤醒后,A去执行拿到资源后的操作,B也成功拿到资源并执行。

   因为步骤6的continue,B不需要等待A执行releaseShare()被唤醒,在A获取到资源时同时B也能快速获取到资源,A、B可以同时执行获得资源后的任务

  

 

  以上就是JUC锁:核心类AQS源码详解()的详细内容,想要了解更多 JUC锁:核心类AQS源码详解的内容,请持续关注盛行IT软件开发工作室。

郑重声明:本文由网友发布,不代表盛行IT的观点,版权归原作者所有,仅为传播更多信息之目的,如有侵权请联系,我们将第一时间修改或删除,多谢。

留言与评论(共有 条评论)
   
验证码: