AQS源码分析--jdk1.8

JDK1.8

ArrayList源码分析--jdk1.8
LinkedList源码分析--jdk1.8
HashMap源码分析--jdk1.8
AQS源码分析--jdk1.8
ReentrantLock源码分析--jdk1.8

AbstractQueuedSynchronizer概述

  1. AQS是一个基于FIFO队列，可以用于构建锁或者其他相关同步装置的基础框架。
  2. AQS提供了双向链表。
  3. AQS分为共享模式和独占模式。
  4.AQS基于volatile内存可见性和CAS原子性操作实现线程间通信操作。

AbstractQueuedSynchronizer数据结构

  数据结构是集合的精华所在，数据结构往往也限制了集合的作用和侧重点，了解各种数据结构是我们分析源码的必经之路。
  AQS的数据结构如下：双向链表
  
  AQS实现共享资源的访问控制基础：
     1.state字段，即同步器状态字段。用于共享资源的访问控制
     2.CLH队列，FIFO等待队列，存放竞争失败的线程。通常CLH队列是一个自旋队列，AQS以阻塞的方式实现
     CLH队列的使用：

CLH扫盲

自旋锁
学习了解自旋锁之前先回顾一下互斥锁 
互斥锁 
线程在获取互斥锁的时候，如果发现锁已经被其它线程占有，那么线程就会惊醒休眠，然后在适当的时机（比如唤醒）在获取锁。 
自旋锁 
那么自旋锁顾名思义就是“自旋”。就是当一个线程在尝试获取锁失败之后，线程不会休眠或者挂起，而是一直在循环检测锁是否被其它线程释放。 
区别 
互斥锁就是开始开销要大于自旋锁。临界区持锁时间的大小并不会对互斥锁的开销造成影响，而自旋锁是死循环检测，加锁全程消耗cpu，起始开销虽然低于互斥锁，但是随着持锁时间，加锁的开销是线性增长。 
适用的情况 
互斥锁用于临界区持锁时间比较长的操作，比如下面这些情况都可以考虑

临界区有IO操作
临界区代码复杂或者循环量大
临界区竞争非常激烈
单核处理器 
自旋锁就主要用在临界区持锁时间非常短且CPU资源不紧张的情况下。当递归调用时有可能造成死锁。 
线程（节点）队列 
了解了自旋锁之后，在学习ReentrantLock的时候，一个线程在等待锁的时候会被封装成一个Node节点，然后加入一个队列中并检测前一个节点是否是头节点，并且尝试获取锁，如果获取锁成功就返回，否则就阻塞。直到上一个节点释放锁并唤醒它。这样看来似乎跟自旋没什么挂钩。这是因为AQS里面的CLH队列是CLH队列锁的一种变形。先来了解一下CLH队列锁
CLH队列锁 
CLH(Craig, Landin, and Hagersten locks): 是一个自旋锁，能确保无饥饿性，提供先来先服务的公平性。 
CLH锁也是一种基于链表的可扩展、高性能、公平的自旋锁，申请线程只在本地变量上自旋，它不断轮询前驱的状态，如果发现前驱释放了锁就结束自旋。http://www.2cto.com/kf/201412/363574.html这篇文章中有比较详细的图解。 
AQS中的CLH队列 
了解了自旋锁与CLH队列锁之后，在学习AQS中的CLH队列就比较简单了。AQS中的CLH队列主要是对CLH队列锁改动了两个地方 
1.节点结构上做出改变。CLH队列锁的节点包含一个布尔类型locked的字段。如果要获取锁，就将这个locked设置为true。然后就不停的轮训前驱节点的locked是否释放了锁（这个过程我们就叫做自旋）。AQS的CLH队列在结构上引入了头节点，尾节点。并且拥有一个前节点与下一个节点的引用。 
2.在等待获取锁的机制上由自旋改成了等待阻塞。
MCS 
MSC与CLH最大的不同并不是链表是显示还是隐式，而是线程自旋的规则不同:CLH是在前趋结点的locked域上自旋等待，而MSC是在自己的 
结点的locked域上自旋等待。正因为如此，它解决了CLH在NUMA系统架构中获取locked域状态内存过远的问题。

AbstractQueuedSynchronizer源码分析

/*
  * 提供了一个基于FIFO队列，可以用于构建锁或者其他相关同步装置的基础框架
 * 双向链表
 */
public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

   /**
     * 无参构造方法
     */
    protected AbstractQueuedSynchronizer() { }
   /**
     * <pre>
     *      +------+  prev +-----+       +-----+
     * head |      | <---- |     | <---- |     |  tail
     *      +------+       +-----+       +-----+
     * </pre>
     */
    static final class Node {
        /** Marker to indicate a node is waiting in shared mode  模式，分为共享与独占 共享模式 */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode 独占模式 */
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled
         * 结点状态 节点watiStatus的值
         * CANCELLED，值为1，终态，该节点被取消由于超时或中断
         * SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark，所以当前节点release或cancels时，必须unpark它的后继节点
         * CONDITION，值为-2，表示当前节点在等待condition，也就是在condition队列中 该节点处于条件队列中，将不会被用于sync queue，直到节点状态被设置为0
         * PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行releaseShared应该被传播到其他节点
         * 值为0，表示当前节点在sync队列中，等待着获取锁
         * */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate
         */
        static final int PROPAGATE = -3;

        /**
         * Status field, taking on only the values:
         *   SIGNAL:     The successor of this node is (or will soon be)
         *               blocked (via park), so the current node must
         *               unpark its successor when it releases or
         *               cancels. To avoid races, acquire methods must
         *               first indicate they need a signal,
         *               then retry the atomic acquire, and then,
         *               on failure, block.
         *   CANCELLED:  This node is cancelled due to timeout or interrupt.
         *               Nodes never leave this state. In particular,
         *               a thread with cancelled node never again blocks.
         *   CONDITION:  This node is currently on a condition queue.
         *               It will not be used as a sync queue node
         *               until transferred, at which time the status
         *               will be set to 0. (Use of this value here has
         *               nothing to do with the other uses of the
         *               field, but simplifies mechanics.)
         *   PROPAGATE:  A releaseShared should be propagated to other
         *               nodes. This is set (for head node only) in
         *               doReleaseShared to ensure propagation
         *               continues, even if other operations have
         *               since intervened.
         *   0:          None of the above
         * 结点状态
         */
        volatile int waitStatus;

        /**
         * 前驱结点
         */
        volatile Node prev;

        /**
         * 后继结点
         */
        volatile Node next;

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

        /**
         * 下一个等待者
         */
        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;
        }

        // 无参构造函数
        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;
        }
    }

    /**
     * CLH队列中头结点
     */
    private transient volatile Node head;

    /**
     * CLH队列中尾结点
     */
    private transient volatile Node tail;

    /**
     * 同步状态
     * 多线程同步获取资源成功，则state字段会自增；若有线程释放资源，则state字段自减。
     * 信号量  记录该线程持有锁的次数。 该线程每次释放所 信号量 -1。 信号量为零 代表 锁被真正释放
     */
    private volatile int state;

    /**
     * @return current state value
     */
    protected final int getState() {
        return state;
    }

    /**
     * @param newState the new state value
     */
    protected final void setState(int newState) {
        state = newState;
    }

    /**
     * 使用unsafe的cas比较并且交换，保证原子性
     */
    protected final boolean compareAndSetState(int expect, int update) {
        // See below for intrinsics setup to support this
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

AbstractQueuedSynchronizer继承和实现分析

   AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer。
   1. AbstractOwnableSynchronizer是一个抽象类一个同步器，它可能由线程独占。该类为创建可能包含所有权概念的锁和相关同步器提供了基础，但是，子类和工具可以使用适当维护的值来帮助控制和监视访问并提供诊断，实现了Serializable接口，定义了独占模式，设置和获取独占模式下的线程Thread信息。
   2.AbstractOwnableSynchronizer实现了Serializable接口。
     1)Serializable接口，序列化接口，表明该类可以被序列化，什么是序列化？简单的说，就是能够从类变成字节流传输，反序列化，就是从字节流变成原来的类。
   3. AbstractOwnableSynchronizer是一个抽象父类，子类有AbstractQueuedSynchronizer和AbstractQueuedLongSynchronizer，它们2个之间的区别就是异常将所有与状态相关的参数和结果定义为long类型而不是int类型，在创建同步器（例如多级锁和需要64位状态的障碍）时，此类可能很有用。  

AbstractQueuedSynchronizer核心方法分析

1. acquire方法--独占模式

     1)acquire(int arg)；
     以独占模式获取资源，如果获取成功，直接返回，否则进去CLH等待队列，通过自旋知道获取到资源为止，过程中忽略线程中断，获取资源后才进行自我中断(补上)，下面看源码：

/**
 * AQS的独占模式--互斥
 * tryAcquire()尝试直接去获取资源，如果成功则直接返回；
 * addWaiter()将该线程加入等待队列的尾部，并标记为独占模式；
 * acquireQueued()使线程在等待队列中获取资源，一直获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。
 * 如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。
 */
public final void acquire(int arg) {
    if (!tryAcquire(arg) && // 再次尝试上锁 回到了  NonfairSync.tryAcquire 方法， tryAcquire 调用了 Sync.nonfairTryAcquire方法
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 链表尾部添加节点 为独占模式
        selfInterrupt();
}

     2)boolean tryAcquire(int arg)；
     尝试以独占的方式获取资源，成功true，失败false，该方法可以用于实现Lock中的tryLock()方法。

/**
 * tryAcquire尝试以独占的方式获取资源，如果获取成功，则直接返回true，否则直接返回false。该方法可以用于实现Lock中的tryLock()方法。
 * 该方法的默认实现是抛出UnsupportedOperationException，具体实现由自定义的扩展了AQS的同步类来实现。AQS在这里只负责定义了一个公共的方法框架。
 * 这里之所以没有定义成abstract，是因为独占模式下只用实现tryAcquire-tryRelease，而共享模式下只用实现tryAcquireShared-tryReleaseShared。
 * 如果都定义成abstract，那么每个模式也要去实现另一模式下的接口
 * 由子类选择性实现
 */
protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}

     3)Node addWaiter(Node mode)；
     将一个Node节点放入到CLH队列的队尾。

/**
 * 将一个Node节点放入到CLH队列的队尾。
 * 第一步：首先将oldTail赋值给newNode.prev：node.prev = pred, 把当前tail节点赋值到mode新节点的prev前一个,
 * 第二步：将tail赋值给newNode：compareAndSetTail(pred, node) 把当前tail节点的内存地址修改为(指向)新的mode节点，
 * 第三步：将oldTail的next指针指向newNode(即tail)：pred.next = node 把当前tail节点的next后一个赋值为新的mode节点(即tail)
 * 如果队列为空，通过enq(node)方法初始化一个等待队列，并返回当前节点
 */
private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    //尝试快速入队，失败则使用enq()方式
    Node pred = tail;
    if (pred != null) { // 列队尾部不为空
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 列队尾部为空 或者  CAS 操作失败
    enq(node);
    return node;
}

     4)boolean acquireQueued(final Node node, int arg)；
     使线程在等待队列中获取资源，一直获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。

/**
 * 若node节点的前继节点是head节点，则会再次调用tryAcquire()获取资源
 * 判断当前节点的前继节点是否为head节点。若是，则表示该节点有资格尝试获取共享资源。此处的head节点的判断在一定程度上保证资源竞争的公平性
 * shouldParkAfterFailedAcquire()：判断当前节点是否可以安全进入park()
 * parkAndCheckInterrupt()：让线程进入等待
 * 用于队列中的线程自旋地以独占且不可中断的方式获取同步状态（acquire），直到拿到锁之后再返回。该方法的实现分成两部分：
 * 如果当前节点已经成为头结点，尝试获取锁（tryAcquire）成功，然后返回；否则检查当前节点是否应该被park，然后将该线程park并且检查当前线程是否被可以被中断
 */
final boolean acquireQueued(final Node node, int arg) {
    //标记是否成功拿到资源，默认false
    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;
            }
            //判断当前节点是否可以进入park，若可以，让线程进入等待
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        //如果获取资源失败，则取消
        if (failed)
            cancelAcquire(node);
    }
}

     5)void selfInterrupt();
     中断当前线程

 /**
 * Convenience method to interrupt current thread.
 * 中断当前线程
 */
static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

     6)Node enq(final Node node)；
     将当前节点插入等待队列

/**
 * 进行自旋入队方式的enq()方法，基本和addWaiter()方法一致：
 * 用于将当前节点插入等待队列，如果队列为空，则初始化当前队列。整个过程以CAS自旋的方式进行，直到成功加入队尾为止
 */
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        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;
            }
        }
    }
}

     7)boolean compareAndSetHead(Node update)；
     通过原子(CAS)操作 改变上锁状态

/**
 * CAS head field. Used only by enq.
 * 通过原子操作 改变上锁状态
 * this == null
 * 第一个参数为需要改变的对象，第二个为偏移量(即之前求出来的valueOffset的值)，第三个参数为期待的值，第四个为更新后的值
 */
private final boolean compareAndSetHead(Node update) {
    //调用本地方法 实现硬件级别的原子操作 cas
    return unsafe.compareAndSwapObject(this, headOffset, null, update);
}

     8)boolean parkAndCheckInterrupt()；
     让线程去休息，真正进入等待状态
   

 /**
 * 该方法让线程去休息，真正进入等待状态。park()会让当前线程进入waiting状态。在此状态下，有两种途径可以唤醒该线程：
 * 1）被unpark()；2）被interrupt()。需要注意的是，Thread.interrupted()会清除当前线程的中断标记位
 */
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this); // 又是一个底层类 实现线程等待
    return Thread.interrupted();
}

     9)boolean shouldParkAfterFailedAcquire(Node pred, Node node)；
     判断当前节点中的线程，是否可以安全的进入park()。返回true,表示进程可以进入park

/**
 * 该方法的作用在于判断当前节点中的线程，是否可以安全的进入park()。返回true,表示进程可以进入park。若前驱节点的waitStatus为SIGNAL，则表示当前节点可以安全的park()。
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    //如果前驱节点的waitStatus为SIGNAL -1，则表示当前节点可以安全的park()
    if (ws == Node.SIGNAL)
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    // waitStatus>0，即为CANCELLED状态，此时当前节点需要找到状态不为CANCELLED状态的节点，将其设置为自己的前驱节点，并将新的前驱节点的next指向自己。
    // 注意，这样做完之后，那些当前节点的waitStatus状态为CANCELLED的前驱节点链，将成为孤链。但这个孤链仍然有指向原等待队列的prev和next指针。只是原等待队列中已经没有指向孤链的节点指针
    // 将前驱节点移出列队
    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;
    } 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.
         */
        // 走到此处，表明前驱节点的状态为0或PROPAGATE。此时可以将前驱节点的waitStatus设置为SIGNAL状态
        // 注意：这里仍然要返回false，表明当前节点不能被park。我们需要在park之前，重试确认该节点不能获取到资源
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}       

2. acquireShared方法--共享模式

     1)acquireShared(int arg)；
     以共享模式获取资源，如果获取成功，直接返回，否则进去CLH等待队列，通过自旋知道获取到资源为止，过程中忽略线程中断，获取资源后才进行自我中断(补上)，下面看源码：

/**
 *  aqs的共享模式
 *  获取指定量的资源，获取成功则直接返回，获取失败则进入等待队列，直到获取到资源为止，整个过程忽略中断
 */
public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

     2)int tryAcquireShared(int arg)；
     尝试以共享的方式获取资源，成功true，失败false，该方法可以用于实现Lock中的tryLock()方法。

/**
 * tryAcquireShared尝试以共享的方式获取资源，如果获取成功，则直接返回true，否则直接返回false。该方法可以用于实现Lock中的tryLock()方法。
 * 该方法的默认实现是抛出UnsupportedOperationException，具体实现由自定义的扩展了AQS的同步类来实现。AQS在这里只负责定义了一个公共的方法框架。
 * 这里之所以没有定义成abstract，是因为独占模式下只用实现tryAcquire-tryRelease，而共享模式下只用实现tryAcquireShared-tryReleaseShared。
 * 如果都定义成abstract，那么每个模式也要去实现另一模式下的接口
 */
protected int tryAcquireShared(int arg) {
    throw new UnsupportedOperationException();
}

     3)doAcquireShared(int arg)；
     将当前线程加入等待队列尾部休息，直到其他线程释放资源唤醒自己，自己成功拿到相应量的资源后才返回

/**
 * Acquires in shared uninterruptible mode.
 * @param arg the acquire argument
 * 将当前线程加入等待队列尾部休息，直到其他线程释放资源唤醒自己，自己成功拿到相应量的资源后才返回
 */
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) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

     4)boolean parkAndCheckInterrupt()；
     让线程去休息，真正进入等待状态

/**
 * 该方法让线程去休息，真正进入等待状态。park()会让当前线程进入waiting状态。在此状态下，有两种途径可以唤醒该线程：
 * 1）被unpark()；2）被interrupt()。需要注意的是，Thread.interrupted()会清除当前线程的中断标记位
 */
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this); // 又是一个底层类 实现线程等待
    return Thread.interrupted();
}

     5)cancelAcquire(Node node)；
     取消节点

/**
 * 取消节点
 * 列队等待中 抛出异常会调用此方法
 */
private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;

    //找到适合的前继节点，当前节点的waitStatus赋值为CANCELLED
    node.thread = null; // 释放线程

    // Skip cancelled predecessors 前驱节点已被取消  重新定义前驱节点
    Node pred = node.prev;
    //若前继节点是CANCELLED，则继续找前继节点，直至找到一个正常的前继节点赋值给node，作为node的新前继节点
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    Node predNext = pred.next;

    node.waitStatus = Node.CANCELLED; // 取消当前线程 所属的节点（标记为取消），  没有使用 cas  因为 其他线程 不会干扰这里

    // If we are the tail, remove ourselves.
    //特殊情况：node==tail节点，将pred作为tail节点，然后将cancelledNodes节点链从CLH队列剔除
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        int ws;
        //正常情况：则将cancelledNodes节点链从CLH队列剔除
        if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
            pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
            //特殊情况：如果node是head的后继节点，则直接唤醒node的后继节点 pred==head节点：尝试调用unparkSuccessor(node)，尝试唤醒当前节点的后继节点
            unparkSuccessor(node);
        }

        node.next = node; // help GC
    }
}

     6)unparkSuccessor(Node node)；
     唤醒后继节点

    /**
 * 唤醒后继节点
 * 注意：如果当前节点的后继节点为空，或者是被取消的节点。那就从tail节点逆向遍历CLH队列，直至找到一个距离当前节点node最近，且waitStatus<=0的节点，然后唤醒该节点
 */
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; //获取头结点的等待状态
    if (ws < 0) //把该状态设置成0
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * 若后继节点不符合唤醒标准，则逆向遍历CLH，直至找到一个距离当前节点node最近，且waitStatus<=0的节点
     */
    Node s = node.next; //找到后继节点,唤醒后继节点
    if (s == null || s.waitStatus > 0) { //很不巧，后继节点，节点为null,或者被取消
        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); //唤醒节点
}

3. release方法--释放资源

     1)boolean release(int arg)；
     独占模式释放资源

/**
 * 资源的释放
 * 调用tryRelease方法进行释放锁
 * 释放锁成功后，获取头节点，接着唤醒后继节点，调用unparkSuccessor方法
 */
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

     2)boolean tryRelease(int arg)；
     独占模式下尝试释放锁，由子类选择性实现

        protected boolean tryRelease(int arg) {
                throw new UnsupportedOperationException();
        }

     3)boolean releaseShared(int arg)；
     共享模式释放资源

/**
 * Releases in shared mode.  Implemented by unblocking one or more
 * threads if {@link #tryReleaseShared} returns true.
 *
 * @param arg the release argument.  This value is conveyed to
 *        {@link #tryReleaseShared} but is otherwise uninterpreted
 *        and can represent anything you like.
 * @return the value returned from {@link #tryReleaseShared}
 */
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

     4)boolean tryReleaseShared(int arg)；
     共享模式下尝试释放锁，由子类选择性实现

    protected boolean tryReleaseShared(int arg) {
                throw new UnsupportedOperationException();
        }

     5)int fullyRelease(Node node)；
     使用当前节点状态调用release,成功返回状态，失败跑出异常

/**
 * 使用当前节点状态调用release,成功返回状态，失败跑出异常
 */
final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

4. CAS操作

 /**
 * Unsafe类实例
 */
private static final Unsafe unsafe = Unsafe.getUnsafe();
/**  state内存偏移地址 */
private static final long stateOffset;
/**  head内存偏移地址 */
private static final long headOffset;
/** tail内存偏移地址 */
private static final long tailOffset;
private static final long waitStatusOffset;
/** next内存偏移地址 */
private static final long nextOffset;

//静态初始化块
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); }
}

/**
 * CAS head field. Used only by enq.
 * 通过原子操作 改变上锁状态
 * this == null
 * 第一个参数为需要改变的对象，第二个为偏移量(即之前求出来的valueOffset的值)，第三个参数为期待的值，第四个为更新后的值
 */
private final boolean compareAndSetHead(Node update) {
    //调用本地方法 实现硬件级别的原子操作 cas
    return unsafe.compareAndSwapObject(this, headOffset, null, update);
}

/**
 * CAS tail field. Used only by enq.
 */
private final boolean compareAndSetTail(Node expect, Node update) {
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

/**
 * CAS waitStatus field of a node.
 */
private static final boolean compareAndSetWaitStatus(Node node,
                                                     int expect,
                                                     int update) {
    return unsafe.compareAndSwapInt(node, waitStatusOffset,
                                    expect, update);
}

/**
 * CAS next field of a node.
 */
private static final boolean compareAndSetNext(Node node,
                                               Node expect,
                                               Node update) {
    return unsafe.compareAndSwapObject(node, nextOffset, expect, update);
}

5. Condition条件队列

内部类ConditionObject，它实现了Condition接口，主要用于实现条件锁。
ConditionObject中也维护了一个队列，这个队列主要用于等待条件的成立，当条件成立时，其它线程将signal这个队列中的元素，将其移动到CLH的队列中，等待占有锁的线程释放锁后被唤醒。
Condition典型的运用场景是在BlockingQueue中的实现，当队列为空时，获取元素的线程阻塞在notEmpty条件上，一旦队列中添加了一个元素，将通知notEmpty条件，将其队列中的元素移动到AQS队列中等待被唤醒。

/**
 * 构造一个条件队列，来等待条件是否为真
 */
public class ConditionObject implements Condition, java.io.Serializable {
    /** 版本号 */
    private static final long serialVersionUID = 1173984872572414699L;
    /** First node of condition queue. condition队列的头结点 */
    private transient Node firstWaiter;
    /** Last node of condition queue. condition队列的尾结点 */
    private transient Node lastWaiter;

    /**
     * Creates a new {@code ConditionObject} instance.
     * 构造函数
     */
    public ConditionObject() { }

    /**
     * Adds a new waiter to wait queue.
     * @return its new wait node
     * 添加新的waiter到wait队列
     */
    private Node addConditionWaiter() {
        //保存尾结点
        Node t = lastWaiter;
        // If lastWaiter is cancelled, clean out. 尾结点不为空，并且尾结点的状态不为CONDITION
        if (t != null && t.waitStatus != Node.CONDITION) {
            //清除状态为CONDITION的结点
            unlinkCancelledWaiters();
            //将最后一个结点重新赋值给t
            t = lastWaiter;
        }
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null)
            firstWaiter = node;
        else
            t.nextWaiter = node;
        lastWaiter = node;
        return node;
    }

AQS总结

1）AQS分为独占锁和共享锁。
2）AQS分为CLH自旋队列和Condition条件队列。
3）AQS是一个双向链表，由state状态控制。
4）AQS由volatile修饰保证多线程可见，采用CAS保证原子性。

原文链接:https://www.cnblogs.com/hackerxian/p/11328823.html
如有疑问请与原作者联系

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有