Java队列同步器（AQS）究竟是怎么一回事

时间  2019-11-07
标签 java 队列 同步 aqs 究竟 怎么 一回事 栏目 Java 繁體版
原文   原文链接

0 简介

队列同步器AbstractQueuedSynchronizer（后面简称AQS）是实现锁和有关同步器的一个基础框架。java

在JDK5中，Doug Lea在并发包中加入了大量的同步工具，例如重入锁（ReentrantLock）、读写锁（ReentrantReadWriteLock）、信号量（Semaphore）、CountDownLatch等，都是基于AQS的。node

其内部经过一个被标识为volatile的名为state的变量来控制多个线程之间的同步状态。多个线程之间能够经过AQS来独占式或共享式的抢占资源。编程

基于AQS，能够很方便的实现Java中不具有的功能。安全

例如，在锁这个问题上，Java中提供的是synchronized关键字，用这个关键字能够很方便的实现多个线程之间的同步。但这个关键字也有不少缺陷，好比：微信

  • 他不支持超时的获取锁，一个线程一旦没有从synchronized上获取锁，就会卡在这里，没有机会逃脱。因此一般由synchronized形成的死锁是无解的。
  • 不可响应中断。
  • 不能尝试获取锁。若是尝试获取时没获取到，马上返回，synchronized不具有这一特性。

而ReentrantLock基于AQS将上述几点都作到了。并发

1 核心结构

从AbstractQueuedSynchronizer的名字能够看出，AQS中必定是基于队列实现的（Queue）。在AQS内部，是经过链表实现的队列。链表的每一个元素是其内部类Node的一个实现。而后AQS经过实例变量head指向队列的头，经过实例变量tail指向队列的尾。框架

其源码定义以下：工具

/** * 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;


static final class Node {

	/** 标识为共享式 */
    static final Node SHARED = new Node();
    /** 标识为独占式 */
    static final Node EXCLUSIVE = null;

	/** 同步队列中等待的线程等待超时或被中断，须要从等待队列中取消等待，进入该状态的节点状态将再也不变化 */
    static final int CANCELLED =  1;

    /** 当前节点的后继节点处于等待状态，且当前节点释放了同步状态，须要经过unpark唤醒后继节点，让其继续运行 */
    static final int SIGNAL    = -1;

    /** 当前节点等待在某一Condition上，当其余线程调用这个Conditino的signal方法后，该节点将从等待队列恢复到同步队列中，使其有机会获取同步状态 */
    static final int CONDITION = -2;

    /** 表示下一次共享式同步状态获取状态将无条件的传播下去 */
    static final int PROPAGATE = -3;

	/* 当前节点的等待状态，取值为上述几个常量之一，另外，值为0表示初始状态 */
    volatile int waitStatus;

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

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

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

    /* 等待队列中的后继节点 */
    Node nextWaiter;
    
    // ...
}
复制代码

当线程经过AQS获取同步状态时，AQS会将当前线程封装到Node内部，并入队。因此在多个线程并发获取同步状态时，AQS内部会持有以下结构的队列：ui

AQS内部队列

下文会基于这个队列模型，说明一下线程在AQS中获取同步状态时的原理。this

2 实现原理

从AQS的名字能够看出来，做者是但愿AQS做为一个基类来向外提供服务的（以Abstract标识）。因此一般AQS是以继承的方式使用的。

AQS提供了几个模板方法供实现类本身实现定制功能。

这几个方法是：

  • boolean tryAcquire(int arg)：独占式的获取同步状态，一般经过以CAS的方式修改state的值来实现特定功能。
  • boolean tryRelease(int arg)：独占式的释放同步状态，一般也是修改state的值。
  • int tryAcquireShared(int arg)：共享式的获取同步状态，返回值>=0表示成功，不然失败。
  • boolean tryReleaseShared(int arg)：共享式的释放同步状态，一样经过修改state值来实现。
  • boolean isHeldExclusively()：表示AQS是否被当前线程独占。

这几个方法的默认实现都会抛出UnsupportedOperationException异常。

目前咱们不用关心这几个方法，只要明白其内部是经过控制state的值来管理同步状态便可。

2.1 同步状态的获取

一般，实现类会优先尝试修改state的值，来获取同步状态。例如，若是某个线程成功的将state的值从0修改成1，表示成功的获取了同步状态。 这个修改的过程是经过CAS完成的，因此能够保证线程安全。

反之，若是修改state失败，则会将当前线程加入到AQS的队列中，并阻塞线程。

AQS内部提供了三个方法来修改state的状态，其源码以下：

/** * Returns the current value of synchronization state. * This operation has memory semantics of a {@code volatile} read. * @return current state value */
protected final int getState() {
    return state;
}

/** * Sets the value of synchronization state. * This operation has memory semantics of a {@code volatile} write. * @param newState the new state value */
protected final void setState(int newState) {
    state = newState;
}

/** * Atomically sets synchronization state to the given updated * value if the current state value equals the expected value. * This operation has memory semantics of a {@code volatile} read * and write. * * @param expect the expected value * @param update the new value * @return {@code true} if successful. False return indicates that the actual * value was not equal to the expected value. */
protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
复制代码

2.2同步队列

如上文所述，AQS内部其实是一个FIFO的双端队列，当线程获取同步状态失败时，就会构建一个Node并添加到队列尾部（此过程是线程安全的，CAS实现），并阻塞当前线程（经过LockSupport.park()方法）； 当释放同步状态时，AQS会先判断head节点是否为null，若是不是null，说明有等待同步状态的线程，就会尝试唤醒head节点，使其从新竞争同步状态。

2.3 独占式同步状态的获取

独占式的意思就是说同一时间只能有一个线程得到同步状态。

AQS会先尝试调用实现类的tryAcquire方法获取同步状态，若是获取失败，会尝试将其封装为Node节点添加到同步队列尾部。

独占式同步状态的获取经过AQS的acquire方法实现。其源码以下：

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

这个方法会先尝试获取一次同步状态（tryAcquire），若是获取失败，会经过addWaiter方法将当前线程加入到同步队列。 并在acquireQueued方法中将当前线程阻塞（LockSupport.park()），并进入自旋状态，以获取同步状态。

下面具体看一下他是如何构建Node并将其添加到队尾的。 首先是addWaiter：

/** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared * @return the new node */
private Node addWaiter(Node mode) {
    // mode = Node.EXCLUSIVE，表示是独占模式
    Node node = new Node(Thread.currentThread(), mode);
    // 先快速的经过CAS的方式将Node添加到队尾，若是失败，再进入enq方法经过无限循环添加
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

/** * Inserts node into queue, initializing if necessary. See picture above. * @param node the node to insert * @return node's predecessor */
private Node enq(final Node node) {
    // 无限循环的将node添加到队尾，保证能添加成功
    /* 注意：若是是首次向队列中添加Node，那么调addWaiter方法时，tail仍是null，因此addWaiter方法不会设置成功，会直接进入这个方法 进入这个方法后，因为tail仍然是null，因此会走到第一个if里面，这是会建立一个空的Node出来做为头结点 而后再次循环，此时tail不是null了，会进入else的代码中，这时，才会将须要add的Node添加到队列尾部。 也就是说，首次建立队列时，会默认加一个空的头结点。 */
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
复制代码

再看下acquireQueued方法：

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 进入自旋，不断的获取同步状态
        for (;;) {
            // 获取node在队列中的前驱节点
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                // 若是成功进入到这块代码，说明成功的获取了同步状态
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 获取不成功，调用LockSupport.park()方法将当前线程阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
复制代码

shouldParkAfterFailedAcquire方法用户判断是否须要阻塞当前线程，方法内会操做当前队尾节点的前驱节点的waitStatus，并依据waitStatus判断是否须要park。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL) // Node.SIGNAL == -1
        /* * 代表当前节点须要其余线程的唤醒才能继续执行，此时能够安全的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;
    } else {
        /* * 若是一个节点是初始状态，即waitStatus=0时， * 将前驱节点的waitStatus设置为-1，代表其须要别的线程唤醒才能继续执行 */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
复制代码

当shouldParkAfterFailedAcquire方法判断当前节点须要被park时，会调用parkAndCheckInterrupt将其阻塞：

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}
复制代码

2.4 独占式同步状态的释放

独占式的同步状态释放，在AQS中是经过release()方法实现的。此方法源码以下：

public final boolean release(int arg) {
    // 尝试调用实现类的tryRelease方法来修改同步状态（state）
    if (tryRelease(arg)) {
        Node h = head;
        /* 1.若是head节点是null，表示没有其余线程竞争同步状态，直接返回释放成功 2.若是head节点不是null，代表有竞争。经过unparkSuccessor方法，经过unpark方法唤醒head节点的next节点。使其从新尝试竞争同步状态。 */
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}
复制代码

unparkSuccessor方法会唤醒head节点的next节点，使其能够从新竞争同步状态：

private void unparkSuccessor(Node node) {
    /* * 若是waitStatus的值是负数，例如：-1（等待signal） * 则将其值还原为0 */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /* * 获取头结点的next节点，若是非空，则unpark他 */
    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);
}
复制代码

2.5 独占式同步状态获取与释放-图示

下面会经过画图方式展现一下源码中的过程，首先咱们假设tryAcquire的实现以下：

boolean tryAcquire(int acquires) {
    return compareAndSetState(0, acquires);
}
复制代码

参数acquires固定传1，意为：经过CAS，若是成功将state的值从0修改成1，表示获取同步状态成功，不然失败，须要加入同步队列。

假设tryRelease的实现以下：

boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (c == 0) {
        setState(c);
        return true;
    }
    return false;
}
复制代码

参数releases固定传1，意为：若是当前state-1=0，视为释放成功，其余线程可竞争同步状态。

假设有三个线程并发获取同步状态，标识为t一、t二、t3，三个线程同时经过acquire方法修改state值。

假设t1修改为功，t2和t3修改失败。

t1修改为功以后，将state值变为1，并直接返回。此时head和tail都是空，因此同步队列也是空的。此时同步队列状态以下：

t1成功修改state状态

t2线程竞争同步状态失败，加入到同步队列中：

t2加入同步队列

t3线程竞争同步状态失败，加入到同步队列中：

t3加入同步队列

t1线程执行完毕，释放资源。 先将state还原为0，再unpark头结点的next节点（t2节点），使之重获同步状态的竞争资格。

t1释放同步状态

假设t2被唤醒后成功的获取到了同步状态（即调用tryAcquire方法并成功将state设置为1），t2会将本身所在的Node设置为head节点，并将原head节点的next设置为null（有助于GC）

t2从新获取同步状态

t2执行完成，释放同步状态，将state设置为0，同时唤醒t3，使之再次具有竞争资格

t2释放同步状态

假设t3成功获取同步状态，此时t3将本身所在的Node设置为head节点，并将以前的head节点的next设置为null（即将t2的next设置为null）

t3从新获取同步状态

t3执行完毕，释放同步状态，将state设置为0。因为此时其waitStatus等于0，表示已经没有后继节点须要unpark，直接返回释放成功

t3释放同步状态

最后的t3节点并无被清空，由于他能够用做下一次同步状态竞争的head节点。

2.6 超时获取同步状态

tryAcquireNanos方法实现了这个功能。他与上面描述的获取同步状态的过程大体相同，只不过是加上了时间的判断。 也就是说，每次自旋获取同步状态时，先判断当前时间是否超过了指定的超时时间，若是超时直接返回获取失败。

下面来看下源码，tryAcquireNanos方法源码以下：

public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();

    // 先尝试获取同步状态，若是失败，尝试超时获取
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}
复制代码

能够发现，最终是doAcquireNanos方法实现的超时功能，这个方法中，大部分逻辑与上面的过程是一直的。 注释中说明了有区别的地方。

private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;

    // 计算出超时那个时间点的时间戳
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            // 判断，若是超时，直接返回获取失败
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                return false;
            // 没超时的话，判断剩余时间是否大于1000纳秒，若是大于才park当前线程
            // 不然，不park，直接进入下一次自旋获取，由于这个时间足够小了，可能已经超出了一次系统调用的时间
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold) // spinForTimeoutThreshold = 1000
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
复制代码

3 参考

  • Java并发编程的艺术 方腾飞，魏鹏，程晓明
欢迎关注个人微信公众号

公众号
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程