AQS

一. 基础原理

AQS是基于双向链表结构的队列实现一个同步器。通过多个线程同步竞争state的状态获取锁，未获取到锁的线程放入队列中进行等待。

AQS可以认为是以CLH琐为基础的变种实现

CLH锁也是一种基于链表的可扩展、高性能、公平的自旋锁，申请线程仅仅在本地变量上自旋，它不断轮询前驱的状态，假设发现前驱释放了锁就结束自旋。从实现上看，CLH锁是一种自旋锁，能确保无饥饿性，提供先来先服务的公平性。

AQS的注释中包含很多实现的细节：

• 节点维护线程信息和状态
• 队首节点不一定是最先获取到琐
• 双向链表涉及方式
• 虚拟头尾节点是懒初始化
• Condition队列的实现

1.1 理论功能

1. 同步状态的原子性管理。
2. 等待线程队列的管理。
3. 线程的阻塞与解除阻塞。

伪代码：

// acquire操作如下：
while (synchronization state does not allow acquire) {
    enqueue current thread if not already queued;
    possibly block current thread;
}
dequeue current thread if it was queued;

//release操作如下：
update synchronization state;
if (state may permit a blocked thread to acquire){
    unblock one or more queued threads;
}

---

二、设计理念

模板设计模式是AQS的主要设计方式

命名规则：

1. 锁需要提供获取和释放接口，分别使用aquire和release
2. 支持互斥和共享的锁模式，使用shared表示共享模式
3. 支持中断特性，名称上增加Interruptibly
4. 支持超时特性，方法签名增加时间参数

因此锁的获取提供了6种api操作（2种模式+中断+超时），释放提供了2种api操作（2种锁模式）。 需要说明支持超时的方法同时也支持中断。

通常公共方法为入口，调用用户实现tryXXX的方法进行，如果失败则进入doXXX方法执行内部实现逻辑。如acquire--> tryAcquire-->doAcquire

已经实现的公共方法

//互斥模式
public final void acquire(int arg) {}
public final void acquireInterruptibly(int arg) throws InterruptedException {}
public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {}
public final boolean release(int arg) {}


//共享模式
public final void acquireShared(int arg) {}
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {}
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {}
public final boolean releaseShared(int arg) {}

需要实现的方法

protected boolean tryAcquire(int arg) {}
protected boolean tryRelease(int arg) {}
protected int tryAcquireShared(int arg) {}
protected boolean tryReleaseShared(int arg) {}
protected boolean isHeldExclusively() {}

三、代码实现

3.1 状态管理

状态管理的API相对简单，关键点是使用volatile修饰，保证state的可见性，但不同实现中体现不同的作用和意义。

状态控制使用CAS更新状态，保证操作的原子性

private volatile int state;

protected final int getState() {return state;}
protected final void setState(int newState) {state = newState;}
protected final boolean compareAndSetState(int expect, int update) {
    return STATE.compareAndSet(this, expect, update);
}

3.2 队列管理

队列使用双向链表进行控制，实现线程的阻塞和唤醒操作

1. 节点结构：

volatile Node prev;
volatile Node next;
volatile int waitStatus;
volatile Thread thread;
Node nextWaiter;  // 标记互斥/共享模式

其中，waitStatus可以取值:

static final int CANCELLED =  1;
static final int SIGNAL    = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;

独占模式基本只使用CANCELLED和SIGNAL。

而nextWaiter取值

static final Node SHARED = new Node();
static final Node EXCLUSIVE = null;

独占模式始终为null

需要注意在获取前驱节点时，如果前驱节点时null会抛出NPE，此异常会被在后面被当作取消节点的操作使用。

2. 入队操作

入队涉及到两个方法addWaiter(mode)和enq(node)，

• addWaiter表示尝试快速入队，避免进入自旋循环中。其参数表示节点模式，分互斥和共享两种
• enq以for循环的自旋方式把节点加入到队尾，并且队列的延迟初始化也是在第一个节点入队的时候进行的，初始化之后tail节点就不会为null。

addWaiter通常是入队入口，将线程和节点模式封装称一个节点，入队成功返回该节点。

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

private Node enq(final Node 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;
            }
        }
    }
}

指针顺序：

1. 将tail指向的节点当作前驱节点
2. 当前节点的prev指针指向前驱节点
3. 将tail指向当前节点（CAS保证原子性）
4. 第3步成功后，将前驱节点的next指针执行当前节点

入队逻辑： 如果队列已经初始化，则放到尾节点 如果没有初始化，第一次循环将head和tail初始化（哨兵节点），第二以及以后的循环将节点放置到尾节点，直到成功。

public final boolean hasQueuedPredecessors() {
    // The correctness of this depends on head being initialized
    // before tail and on head.next being accurate if the current
    // thread is first in queue.
    Node t = tail; // Read fields in reverse initialization order
    Node h = head;
    Node s;
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

头节点管理

head节点是哨兵节点，不指定确定线程。 状态设置则有后继节点管理

private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

3.3 线程管理

1. 获取互斥状态

线程以互斥的方式获取状态量，如果成功则返回，如果失败则进入队列进行等待，并且不会响应中断。

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

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) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

子类的tryAcquire成功则直接返回。否则调用addWaiter将当前线程封装称一个队列节点加入队列并返回该节点，acquireQueued根据该节点的前驱节点状态对该节点进行或者阻塞操作。

该节点被unpark后，如果前驱节点是head节点，则尝试获取同步状态，如果获取成功，则替换head节点。替换后方法调用返回，节点实际已经出队，线程不再阻塞，因此setHead节点将属性置空。

acquireQueued返回true表示线程唤醒后已经被中断，需要调用线程处理中断

failed=true有两种情况，即node.predecessor()抛出异常，或者state的值内存溢出。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

1. 前驱节点的状态是SIGNAL，则需要park并等待前驱节点唤醒
2. 前驱节点的状态是ws >0, 表示CANCEL，则需要将已经取消节点移除。
3. 前驱节点的状态是其他情况，将前驱节点的状态设置为SIGNA

此方法返回false后会重新进入循环，如果前驱不是head或者同步器还没有被释放，则重新进入此方法，最终会返回true。

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

将线程阻塞，唤醒后返回线程的中断状态，并清除中断状态。acquireQueued方法中如果收到这个方法返回ture，则表示再阻塞过程中线程进行了中断操作，因此会重新设置中断状态。

如果不清除中断标记，含有中断标记的线程调用park是无效的。

private void cancelAcquire(Node node) {
    if (node == null)
        return;

    node.thread = null;

    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    Node predNext = pred.next;

    node.waitStatus = Node.CANCELLED;

    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        int ws;
        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 {
            unparkSuccessor(node);
        }

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

取消节点会按节点位置分情况处理。

1. tail节点：跳过取消的节点找到新的tail，cas设置tail点，并将tail节点的next节点cas设置为null
2. 中间节点，也不是head的next节点：前一个节点必须有thread，且前一个节点状态为-1或者可以设置为-1，并且后一个节点有效，则通过cas的方式前后节点，删除当前节点
3. head的next节点：则唤醒下一个节点，会进行cancal节点的移除

响应中断的方式获取互斥的状态量

public final void acquireInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (!tryAcquire(arg))
        doAcquireInterruptibly(arg);
}

private void doAcquireInterruptibly(int arg)
    throws InterruptedException {
    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;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

响应中断和不响应中断的区别就是检查到线程被中断时抛出中断异常。

支持超时的获取互斥状态量

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

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;
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

每次被唤醒后计算超时时间，并且大于自旋时间后，阻塞指定的时间。

释放互斥状态量

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

唤醒后继节点

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;
    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);
}

共享式获取

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) {
                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);
    }
}

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);
    
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

共享式和互斥式的区别式setHeadAndPropagate，即获取成功后操作。

propagate的含义：

1. 小于0，获取同步状态失败
2. 大于0，获取同步状态成功，还有剩余的同步状态可用于其他线程获取
3. 等于0，获取同步状态成功，没有剩余的同步状态可用于其他线程获取

支持中断

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

响应中断和不响应中断的区别是是否抛出异常

支持超时

public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquireShared(arg) >= 0 ||
        doAcquireSharedNanos(arg, nanosTimeout);
}

private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
            }
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                return false;
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

释放

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

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

4 Condition

Condition使一个同步对象创建出多个等待集合，替换和丰富了Object对象提供给的wait/notify方法。

Condition通过实现一个将线程封装成节点后，组成一个单向队列进行等待

节点主要信息

当前节点的线程 thread
当前节点的状态 waitStatus
当前节点的下一个节点指针 nextWaiter

在条件队列中，我们只需要关注一个值即可那就是CONDITION。它表示线程处于正常的等待状态，而只要waitStatus不是CONDITION，我们就认为线程不再等待了，此时就要从条件队列中出队。

条件队列和同步队列节点的区分方法：

同步队列中只使用prev、next来串联链表，而不使用nextWaiter;
条件队列中只使用nextWaiter来串联链表，而不使用prev、next.
这样就使用同样的Node数据结构的创建完全独立的两种链表。

中断检查

LockSupport.park(this)方法返回后，会进行中断检查。 方法返回的原因主要有三个，即中断，unpark或者其他原因，所以要检查中断，如果是中断则通过break退出，如果是唤醒则已经进入同步队列而退出。

interruptMode取三个值

• 0： 代表整个过程中一直没有中断发生。
• THROW_IE ： 表示退出await()方法时需要抛出InterruptedException，这种模式对应于中断发生在signal之前
• REINTERRUPT ： 表示退出await()方法时只需要再自我中断以下，这种模式对应于中断发生在signal之后。

private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}
final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        enq(node);
        return true;
    }
    while (!isOnSyncQueue(node))
        Thread.yield();
    return false;
}

一个调用了await方法挂起的线程在被中断后不会立即抛出InterruptedException，而是会被添加到同步队列中去争锁，如果争不到，还是会被挂起；

只有争到了锁之后，该线程才得以从同步队列和条件队列中移除，最后抛出InterruptedException。

中断对它意义更多的是体现在将它从条件队列中移除，加入到同步队列中去争锁，

await区别：

await() 中断后进入同步队列
awaitUninterruptibly()  中断后仍在条件队列

参考资料

1. 并发条件队列之Condition 精讲 (opens new window)