Java并发编程java 底层

Java并发编程——AQS源码解析

2018-06-12  本文已影响67人  Ant_way

Java并发编程——AQS源码解析

一、AQS是什么？

AQS是一个基于先进先出（FIFO）等待队列的实现阻塞锁和同步器的框架。AQS通过一个volatile int state变量来保存锁的状态。子类必须通过：

三个方法修改获取锁的状态值state。

CAS (compare and swap) 比较并交换，就是将内存值与预期值进行比较，如果相等才将新值替换到内存中，并返回true表示操作成功；如果不相等，则直接返回false表示操作失败。CAS操作大多都是靠CPU原语来实现。CAS操作经常被用来实现无锁数据结构，在java.util.concurrent包中就有很多这样的数据结构：ConcurrentLinkedQueue、ConcurrentLinedDeque、ConcurrentHashMap、ConcurrentSkipListMap、ConcurrentSkipListSet。

1、AQS支持的锁的类别
AQS支持独占锁和共享锁两种。

2、基于AQS实现锁
AQS中没有实现任何的同步接口，所以一般子类通过继承AQS以内部类的形式实现锁机制。一般通过继承AQS类实现同步器，通过getState、setState、compareAndSetState来监测状态，并重写以下方法：

一般来说，自定义同步器要么是独占方法，要么是共享方式，他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式，如ReentrantReadWriteLock。

3、AQS同步器的存储结构

二、AQS源码解析

1、存储节点Node
我们一直在说AQS是基于FIFO队列的存储结构，它是以内部类Node节点的形式进行存储。这个等待队列是CLH同步队列。

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;
    /** waitStatus value to indicate thread is waiting on condition */
    static final int CONDITION = -2;
    /**
     * 下一次共享式同步状态获取将会无条件地传播下去
     */
    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
     *
     * The values are arranged numerically to simplify use.
     * Non-negative values mean that a node doesn't need to
     * signal. So, most code doesn't need to check for particular
     * values, just for sign.
     *
     * The field is initialized to 0 for normal sync nodes, and
     * CONDITION for condition nodes.  It is modified using CAS
     * (or when possible, unconditional volatile writes).
     */
    volatile int waitStatus;

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

    /**
     * 后驱节点
     */
    volatile Node next;

    /**
     * 获取同步状态的线程
     */
    volatile Thread thread;

    /**
     * Link to next node waiting on condition, or the special
     * value SHARED.  Because condition queues are accessed only
     * when holding in exclusive mode, we just need a simple
     * linked queue to hold nodes while they are waiting on
     * conditions. They are then transferred to the queue to
     * re-acquire. And because conditions can only be exclusive,
     * we save a field by using special value to indicate shared
     * mode.
     */
    Node nextWaiter;

    /**
     * Returns true if node is waiting in shared mode.
     */
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

    /**
     * Returns previous node, or throws NullPointerException if null.
     * Use when predecessor cannot be null.  The null check could
     * be elided, but is present to help the VM.
     *
     * @return the predecessor of this node
     */
    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内部类中，声明了pre、next节点用于队列的连接，同时保存了waitStatus状态。

2、AQS源码解析
在面向对象的世界中，想要了解一个类有什么特点，就要看它的属性。通过查看源码，我们看到AQS类包含：

/**
 * Head of the wait queue, lazily initialized.  Except for
 * initialization, it is modified only via method setHead.  Note:
 * If head exists, its waitStatus is guaranteed not to be
 * CANCELLED.
 */
private transient volatile Node head;

/**
 * Tail of the wait queue, lazily initialized.  Modified only via
 * method enq to add new wait node.
 */
private transient volatile Node tail;

/**
 * The synchronization state.
 */
private volatile int state;

AQS基于FIFO队列，接下来就依照acquire-release、acquireShared-releaseShared的次序来分析入队和出队。

1、acquire(int)
此方法是独占模式下线程获取共享资源的顶层入口。如果获取到资源成功，线程直接返回，否则进入等待队列，直到获取到资源为止，且整个过程忽略中断的影响。这也正是lock()的语义，当然不仅仅只限于lock()。获取到资源后，线程就可以去执行其临界区代码了。下面是acquire源码：

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

在acquire方法中调用了tryAcquire()、acquireQueued()、addWaiter()三个方法。

首先通过tryAcquire方法尝试申请独占锁。如果获取成功则返回true，否则返回false。

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

呦西，源码中直接throw一个异常。结合我们前面自定义锁的知识，AQS只是一个框架，具体资源获取和释放方式交由自定义同步器实现。AQS这里只定义了一个接口，具体资源的获取交由自定义同步器去实现了（通过state的get/set/CAS）！！！至于能不能重入，能不能加塞，那就看具体的自定义同步器怎么去设计了！！！当然，自定义同步器在进行资源访问时要考虑线程安全的影响。

这里之所以没有定义成abstract，是因为独占模式下只用实现tryAcquire-tryRelease，而共享模式下只用实现tryAcquireShared-tryReleaseShared。如果都定义成abstract，那么每个模式也要去实现另一模式下的接口。说到底，Doug Lea还是站在咱们开发者的角度，尽量减少不必要的工作量。

接着就是addWaiter()方法用于将当前线程添加到等待队列的队尾，并返回当前线程所在的节点。

private Node addWaiter(Node mode) {
    //以给定的Node节点模式构建当前线程的Node节点，在acquire方法中传入的是EXCLUSIVE独占式节点
    Node node = new Node(Thread.currentThread(), mode);
    // 将尾节点进行保存
    Node pred = tail;
    if (pred != null) {//如果尾节点不为null
        //将尾节点设置尾新节点的prev节点
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {//通过CAS保证，确保节点能够被线程安全的添加
            //将当前界定指向前驱的next节点
            pred.next = node;
            return node;
        }
    }
    //如果尾节点为null，则通过enq进行入队
    enq(node);
    return node;
}

//同步器通过死循环的方式来保证节点的正确添加，在“死循环” 中通过CAS将节点设置成为尾节点之后，
//当前线程才能从该方法中返回，否则当前线程不断的尝试设置。
private Node enq(final Node node) {
    //CAS"自旋"，直到成功加入队尾
    for (;;) {
        //尾节点临时存储
        Node t = tail;
        if (t == null) { // Must initialize
            //如果tail为null，则将
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

在addWaiter(Node node)方法中，将当前线程节点添加到等待队列中。

enq.png

acquireQueued在队列中的线程获取锁

/**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*
* @param node the node
* @param arg the acquire argument
* @return {@code true} if interrupted while waiting
* 
* acquireQueued方法当前线程在死循环中获取同步状态，而只有前驱节点是头节点才能尝试获取同步状态（锁）（ p == head && tryAcquire(arg)）
*     原因是:1.头结点是成功获取同步状态（锁）的节点，而头节点的线程释放了同步状态以后，将会唤醒其后继节点，后继节点的线程被唤醒后要检查自己的前驱节点是否为头结点。
*           2.维护同步队列的FIFO原则，节点进入同步队列之后，就进入了一个自旋的过程，每个节点（或者说是每个线程）都在自省的观察。
* 
*/
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;
            }
            // 否则检查前一个节点的状态，看当前获取锁失败的线程是否要挂起
            if (shouldParkAfterFailedAcquire(p, node) &&
                //如果需要挂起，借助JUC包下面的LockSupport类的静态方法park挂起当前线程，直到被唤醒
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        //如果有异常
        if (failed)
            //取消请求，将当前节点从队列中移除
            cancelAcquire(node);
    }
}

通过addWaiter方法添加到等待队列中后，在通过acquireQueued方法进行锁的获取

函数流程：

独占式锁获取流程
调用同步器的acquire(int arg)方法可以获取同步状态，该方法对中断不敏感，即线程获取同步状态失败后进入同步队列，后续对线程进行中断操作时，线程不会从同步队列中移除。获取流程：

  1. 当前线程通过tryAcquire()方法尝试获取锁，成功则直接返回，失败则进入队列排队等待，通过CAS获取同步状态。
  2. 如果尝试获取锁失败的话，构造同步节点（独占式的Node.EXCLUSIVE），通过addWaiter(Node node,int args)方法,将节点加入到同步队列的队列尾部。
  3. 最后调用acquireQueued(final Node node, int args)方法，使该节点以死循环的方式获取同步状态，如果获取不到，则阻塞节点中的线程。acquireQueued方法当前线程在死循环中获取同步状态，而只有前驱节点是头节点的时候才能尝试获取锁（同步状态）（ p == head && tryAcquire(arg)）。
lock.png

2、release(int)：独占锁的释放
在AQS中通过release方法进行锁的释放。

public final boolean release(int arg) {
   //调用tryRelease方法释放
    if (tryRelease(arg)) {//如果释放成功
        Node h = head;
        //如果头节点不为null，并且头结点的waitStatus值不为0，即有状态
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
   //没有释放成功返回false
    return false;
}

// tryRelease() 尝试释放当前线程的同步状态（锁）
protected final boolean tryRelease(int releases) {
        //c为释放后的同步状态
      int c = getState() - releases;
      //判断当前释放锁的线程是否为获取到锁（同步状态）的线程，不是抛出异常（非法监视器状态异常）
      if (Thread.currentThread() != getExclusiveOwnerThread())
          throw new IllegalMonitorStateException();
      boolean free = false;
      //如果锁（同步状态）已经被当前线程彻底释放，则设置锁的持有者为null，同步状态（锁）变的可获取
      if (c == 0) {
          free = true;
          setExclusiveOwnerThread(null);
      }
      setState(c);
      return free;
  }

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)
        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;
    //从队列尾部开始往前去找最前面的一个waitStatus小于0的节点。
    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);
}

release()是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源，如果彻底释放了（即state=0）,它会唤醒等待队列里的其他线程来获取资源。

3、acquireShared(int)
此方法是共享模式下线程获取共享资源的顶层入口。它会获取指定量的资源，获取成功则直接返回，获取失败则进入等待队列，直到获取到资源为止，整个过程忽略中断。下面是acquireShared()的源码：

public final void acquireShared(int arg) {
     if (tryAcquireShared(arg) < 0)
         doAcquireShared(arg);
}

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) {//如果到head的下一个，因为head是拿到资源的线程，此时node被唤醒，很可能是head用完资源来唤醒自己的
                int r = tryAcquireShared(arg);//尝试获取资源
                if (r >= 0) {//成功
                    setHeadAndPropagate(node, r);//将head指向自己，还有剩余资源可以再唤醒之后的线程
                    p.next = null; // help GC
                    if (interrupted)//如果等待过程中被打断过，此时将中断补上。
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            
            //判断状态，寻找安全点，进入waiting状态，等着被unpark()或interrupt()
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

这里tryAcquireShared()依然需要自定义同步器去实现。但是AQS已经把其返回值的语义定义好了：负值代表获取失败；0代表获取成功，但没有剩余资源；正数表示获取成功，还有剩余资源，其他线程还可以去获取。所以这里acquireShared()的流程就是：

  1. tryAcquireShared()尝试获取资源，成功则直接返回；
  2. 失败则通过doAcquireShared()进入等待队列park()，直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

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

4、releaseShared()
releaseShared()是共享模式下线程释放共享资源的顶层入口。它会释放指定量的资源，如果成功释放且允许唤醒等待线程，它会唤醒等待队列里的其他线程来获取资源。

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {//尝试释放资源
        doReleaseShared();//唤醒后继结点
        return true;
    }
    return false;
}

此方法的流程也比较简单，一句话：释放掉资源后，唤醒后继。跟独占模式下的release()相似，但有一点稍微需要注意：独占模式下的tryRelease()在完全释放掉资源（state=0）后，才会返回true去唤醒其他线程，这主要是基于独占下可重入的考量；而共享模式下的releaseShared()则没有这种要求，共享模式实质就是控制一定量的线程并发执行，那么拥有资源的线程在释放掉部分资源时就可以唤醒后继等待结点。例如，资源总量是13，A（5）和B（7）分别获取到资源并发运行，C（4）来时只剩1个资源就需要等待。A在运行过程中释放掉2个资源量，然后tryReleaseShared(2)返回true唤醒C，C一看只有3个仍不够继续等待；随后B又释放2个，tryReleaseShared(2)返回true唤醒C，C一看有5个够自己用了，然后C就可以跟A和B一起运行。而ReentrantReadWriteLock读锁的tryReleaseShared()只有在完全释放掉资源（state=0）才返回true，所以自定义同步器可以根据需要决定tryReleaseShared()的返回值。

总结
本节中针对AQS进行了简单的学习，总结中参照了以下文章，推荐大家去阅读下。

上一篇 下一篇

猜你喜欢

热点阅读