JUC锁:核心类AQS源码详解

1 疑点todo和解疑

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

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

JUC锁:核心类AQS源码详解插图

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

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

只有将前置节点状态改为SIGNAL,才能确保当前节点可以被前置unPark唤醒。也就是说阻塞自己前先保证一定能够被唤醒。因为代码中:
独占模式下,唤醒后继前先限制:h.waitStatus != 0
共享模式下,唤醒后继前先限制:h.waitStatus=SIGNAL

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

 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 "understand" 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. 独占模式: 只能够一个线程占有资源,其它尝试获取资源的线程将会进入到队列等待。
  3. 响应中断并终止线程只要被中断就不会获取资源:两种情况的中断: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的分析。

  1. 每一个节点都是由前一个节点唤醒
  2. 当节点发现前驱节点是head并且尝试获取成功,则会轮到该线程运行。
  3. condition queue中的节点向sync queue中转移是通过signal操作完成的。
  4. SIGNAL,表示后面的节点需要运行。
  5. 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);
  1. addWaiter() 或者enq()插入节点时,都是先更新节点的prev域,再更新它前驱的next域。那么通过node.next()取到的后继,后继的prev域一定是指向node本身。如果先更新next域,在更新prev域时出现异常,那么通过.next取到不是完整的节点
  2. unparkSuccessor()唤醒后继时,Node s = node.next; 通过.next来获取node的后继,后继的prev一定指向node本身

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

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

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

  1. 共享锁的传播性。
  2. doReleaseShared()中head改变,会循环唤醒head的后继节点。
  3. 线程获取锁失败后入队列并不会立刻阻塞,而是判断是否应该阻塞shouldParkAfterFailedAcquire,如果前继是head,会再给一次机会获取锁。

3 AQS 简介

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

三个基本组件相互协作:

  1. 同步状态的原子性管理;
  2. 线程的阻塞与唤醒;
  3. 队列的管理;

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

3.1 AQS核心思想

  1. 如果请求的共享资源空闲,则将当前请求线程设置为有效工作线程,并且将共享资源设置为锁状态
  2. 设计一套机制:【线程如何阻塞等待以及被唤醒时锁如何分配】?这个机制AQS是用CLH队列锁实现的
  3. CLH队列锁:一个虚拟的双向队列,AQS是将每条请求共享资源的线程封装成一个CLH锁队列的一个节点(Node)来实现锁的分配。【严格的FIFO队列,框架不支持基于优先级的同步
  4. 使用一个int成员变量来表示同步状态,使用volatile修饰保证线程可见性,并使用CAS思想进行值维护。

3.2 对资源的共享方式

两种方式:

  1. Exclusive(独占):只有一个线程能执行。又可分为公平锁和非公平锁:
    1. 公平锁:按照线程在队列中的排队顺序,先到者先拿到锁
    2. 非公平锁:当线程要获取锁时,无视队列顺序直接去抢锁,谁抢到就是谁的
  2. Share(共享):多个线程可同时执行

3.3 AQS数据结构

分析类,首先就要分析底层采用了何种数据结构,抓住核心点进行分析:
JUC锁:核心类AQS源码详解插图1

  1. Sync queue,即同步队列,是双向链表,包括head节点和tail节点,head节点主要用作后续的调度
  2. 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); }
    }
}

说明:

  1. 属性中包含了头节点head,尾节点tail,状态state、自旋时间spinForTimeoutThreshold
  2. AbstractQueuedSynchronizer抽象的属性在内存中的偏移地址,通过该偏移地址,可以获取和设置该属性的值
  3. 同时还包括一个静态初始化块,用于加载内存偏移地址。

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()方法

该函数以独占模式获取(资源),忽略中断
流程如下:
JUC锁:核心类AQS源码详解插图2

源码如下:

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()总结

  1. 先调用tryAcquire(),由子类实现来尝试加锁,如果获取到锁,则线程继续执行;反则,节点加入队列
  2. 调用addWaiter(),将调用线程封装成为一个节点并放入AQS队列。
  3. 调用acquireQueued(),先park阻塞等待,直到被unPark唤醒。
  4. 如果线程被设置中断,那么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()说明

  1. 使用快速添加的方式(失败不重试)创建新节点并添加到往队列尾部,更新tail
  2. 如果队列还没有初始化或者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()方法总结:

  1. 功能:cas+自旋方式将节点插入队列
  2. 如果队列未初始化,先创建头节点head(head不指向任务线程),再将节点插入到队列(当第一个节点被创建后,队列实际有两个节点:head+业务节点)。
  3. 如果队列已经初始化,则直接插入队列

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()总结

  1. 功能:节点进入AQS队列后,先park阻塞等待,直到被unPark唤醒,或者中断唤醒
  2. 找到非取消状态的前驱(取消状态的将会被移出队列并GC),如果前驱是SIGNAL,那么当前节点进入park阻塞,否则,先将前驱改为SIGNAL,再进入park阻塞。
  3. 被unPark唤醒后,判断前驱是头节点且获取到资源(tryAcquire成功),当前节点晋升为头节点。自此,线程获取到锁
  4. 调用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 > 0); // 找到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()总结:

  1. 如果前驱状态是:SIGNAL,返回true。表示当前节点可以安全地unPark()阻塞
  2. 遇到取消的前驱节点,则跳过。这些被取消的节点会从队列中移除并GC
  3. 如果前驱状态不是: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标记
    }

方法总结:

  1. 执行park操作(前提:前驱状态是SIGNAL),在队列中阻塞等待。
  2. 被unPark()唤醒后,返回线程的interrupted中断标识,并且清除interrupted标记

4.5.8 cancelAcquire()方法

什么时候才会执行cancelAcquire?

在lockInterruptibly()会通过抛出中断异常来执行cancelAcquire方法,lock方法过程中则不会执行该代码,作者这么些的意图在于for循环内部如果出现不可控的因素导致产生未知的异常,则会执行cancelAcquire,很明显这属于一种相对偏保守的保险代码。
// 取消获取锁
    private void cancelAcquire(Node node) {
        // Ignore if node doesn't 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) 已经限制了pred一定是

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

  1. 基于对acquire()方法的分析,调用链:addWait()->enq()->acquireQueue()->cancelAcquire(node),进入到cancelAcquire()时,节点node一定已经在队列中,而且它不会是head,并且没有持有锁
  2. 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后,队列变成这样的:
JUC锁:核心类AQS源码详解插图3

发现:

  1. node没有移出队列,因为被后继的prev所引用。
  2. node.next变了,指向了自身,这就能解释为什么unparkSuccessor()是从后往前遍历:因为取消节点的next域指向了自身,所以不能从通过next来遍历,但prev是完整的,所以通过prev来遍历。
  3. 取消节点,暂存在队列中,当后继节点被唤醒,执行shouldParkAfterFailedAcquire后,取消节点的引用链清空,移出队列,最后GC回收

4.5.9 unparkSuccessor()方法

    // 唤醒node节点的后继
    private void unparkSuccessor(Node node) {

        // 获取node节点的等待状态
        int ws = node.waitStatus;
        if (ws  0) { 
            s = null; 
            // 由尾节点向前倒着遍历队列,但不会超过node节点
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus 

unparkSuccessor()总结

  1. 作用:找到有效的后继节点unPark唤醒
  2. 寻找有效后继时从尾往前倒着遍历:因为取消节点的next域指向了自身,所以不能从通过next来遍历
  3. 将发起unPark唤醒的节点(只能是head)状态改为0(意味着在head唤醒后继,到被后继推出队列的期间,状态变为0)

4.5.10 release()方法

以独占模式释放对象,其源码如下:

public final boolean release(int arg) {
        if (tryRelease(arg)) { //如果释放锁成功
            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()总结:

  1. 功能:释放独占锁
  2. 先调用tryRelease()由子类实现释放锁
  3. 如果释放锁成功,然后unPark唤醒后继节点(没有后继就不需要唤醒)

4.5.11 acquireSharedInterruptibly()方法

   //获取共享资源,响应中断
   public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted()) //读取线程中断标记,然后擦除标记
            throw new InterruptedException(); //中断标记为true,抛出中断异常,停止执行
        if (tryAcquireShared(arg) 

获取共享资源流程图:
JUC锁:核心类AQS源码详解插图4

acquireSharedInterruptibly()总结

  1. 共享模式获取对象,响应中断并终止获取
  2. 先调用子类实现获取资源,没有获取到再加队队列等待。

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()总结

  1. 创建节点并插入aqs队列,将前驱状态改为signal,park阻塞,等待unPark唤醒。
  2. 正常唤醒后,无限循环直到前驱是head并且调用子类方法获取共享资源成功,调用setHeadAndPropagate()成为head并进行共享传播
  3. 被中断唤醒、或者循环等待过程发生中断异常,执行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 

满足调用doReleaseShared的条件分析

  1. propagate > 0
    ReentrantReadWriteLock、CountDownLatch 调用tryAcquireShared()返回1进入,满足条件;Semaphore 进入,propagate可能等于0,不满足,继续2
  2. h == null
    h == null 表示旧head变为null,程序没有地方设置head=null,并且这里h引用着head意味着head不会被GC。 因此,h == null不满足条件,继续3 【不知道哪种情况下h==null todo】
  3. h.waitStatus

  • h.waitStatus==1:取消,不能由取消节点唤醒后继,不满足条件

setHeadAndPropagate()总结:

  1. 方法功能:设置 同步队列的head节点,以及触发"传播"操作:
  2. 如果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()总结

  1. 如果头节点状态为signal,那么CAS更新头节点状态为0,成功则调用unparkSuccessor()唤醒后继,失败则重试
  2. 如果头节点状态为0,那么将CAS更新头节点状态为PROPAGTATE ,失败则重试。
  3. 最后如果判断head是否发生变化,有变化则重复1、2,没有变化则方法结束。
  4. PROPAGTATE状态的意义是,增加一个状态判断,当前驱获取资源,后继同时也有机会获取到资源

4.5.15 releaseShared()方法

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

releaseShared()方法总结

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

5 取消节点移出链表分析

有两种情景,会将取消节点彻底移出链表:

  1. 头节点unPark唤醒后继时,后继节点唤醒后重新进入shouldParkAfterFailedAcquire()
  2. 取消节点后面有新节点入列时,新节点执行shouldParkAfterFailedAcquire()

以第一个情景为例子分析:
JUC锁:核心类AQS源码详解插图5

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源码详解插图6

文章来源于互联网:JUC锁:核心类AQS源码详解

THE END
分享
二维码