AQS队列同步器

Node 节点

class Node {
    // static final 修饰的常量
    static final Node SHARED = new Node();
    static final Node EXCLUSIVE = null;

    // 等待状waitStatus的取值
    // static final int INITIAL = 0;
    static final int CANCELLED =  1;
    static final int SIGNAL    = -1;
    // 节点在等待队列中, 等待其它线程调用condition.signal().
    // 当其它线程调用condition.signal()之后, 该节点会从等待队列转移到同步队列中
    static final int CONDITION = -2;
    static final int PROPAGATE = -3;

    // 等待状态
    volatile int waitStatus;

    volatile Node prev;
    volatile Node next;

    volatile Thread thread;

    Node nextWaiter;

    // head和tail对应
    private transient volatile Node head;
    private transient volatile Node tail;

    /**
     * 加锁的数量
     * 对一个线程多次加可重入锁时, 该值会增加
     */
    private volatile int state;

    static final long spinForTimeoutThreshold = 1000L;

    // Unsafe对象
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final long stateOffset;
    private static final long headOffset;
    private static final long tailOffset;
    private static final long waitStatusOffset;
    private static final long nextOffset;

    final boolean isShared() {
        return nextWaiter == SHARED;
    }

    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

    // Used by SHARED
    Node() {    
    }

    // Used by addWaiter
    Node(Thread thread, Node mode) {     
        this.nextWaiter = mode;
        this.thread = thread;
    }

    // Used by Condition
    Node(Thread thread, int waitStatus) { 
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

Condition 对象

Condition 类

public class ConditionObject implements Condition, java.io.Serializable {

    // 等待队列的第一个节点
    private transient Node firstWaiter;
    // 等待队列(condition queue)的最后一个结点
    private transient Node lastWaiter;

    /** Mode meaning to reinterrupt on exit from wait */
    private static final int REINTERRUPT =  1;
    /** Mode meaning to throw InterruptedException on exit from wait */
    private static final int THROW_IE    = -1;

    private static final long serialVersionUID = 1173984872572414699L;

    // 唯一构造器, 默认情况下firstWaiter和lastWaiter都为null
    public ConditionObject() { }
}

addConditionWaiter() 方法（添加到等待队列）

功能：将当前线程封装成Node节点添加到等待队列（condition queue）的末尾

执行流程：

• 首次调用 await() 方法时，等待队列中 firstWaiter 和 lastWaiter 均为 null。将当前线程封装成 Node 后，为 firstWaiter 和 lastWaiter 赋值

• 非首次调用 await() 方法时，TODO

private Node addConditionWaiter() {
    // 1. 首次调用时, lastWaiter为null
    // 2. 非首次调用时, 找到等待队列中的最后一个Node节点
    Node t = lastWaiter;

    // 这段的含义是什么?
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }

    // 为当前调用await()方法的线程创建Node节点
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

release(int arg) 方法（释放锁）

功能：释放指定arg数量的可重入锁

执行流程：

1. 调用tryRelease()方法，释放指定数量的锁
2. 首次调用时，会释放掉全部的锁，此时tryRelease返回true。
3. 非首次调用时，

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        // head并不为null
        Node h = head;
        if (h != null && h.waitStatus != 0){
            // 当前线程即将park()阻塞, 调用unpark()唤醒后续节点
            unparkSuccessor(h);            
        }
        return true;
    }
    return false;
}

// AQS中并没有真正意义上地实现tryRelease(), 而是交给子类
protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
}

// ReentrantLock中对tryRelease()的实现
// tryRelease()的返回值表示当前线程是否释放掉锁, true表示释放掉锁, false表示还没有完全释放掉锁
protected final boolean tryRelease(int releases) {
    // 对Node节点的state值减去指定的数: 释放指定数量的锁
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    // 如果state值为0, 那么
    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;
    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);
}

fullyRelease(Node node) 方法

执行流程：

1. 调用 getState() 方法，获取 node 节点对应线程上可重入锁的次数（state值）
2. 调用 release(savedState) 一次性全部释放掉添加的可重入锁
   • 如果release()释放成功，返回释放掉的可重入锁的数量
   • 如果release()释放失败，将当前线程的等待状态waitStatus设置为CANCELLED

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();
        if (release(savedState)) {
            // 释放成功, 将标志为设置为false, 避免finally中将node的waitStatus状态设置为取消
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

await() 方法

执行流程：

1. 对被中断的线程调用await()方法，产生中断异常

2. 调用addConditionWaiter()方法，将当前线程加入等待队列

3. 调用fullyRelease(node)方法，释放当前线程添加的可重入锁（针对绑定该Condition对象的那把锁，Condition condition = lock.newCondition()）

4. 调用 isOnSyncQueue(node) 方法，判断当前线程是否在同步队列中，对于首次调用的线程来说，必定返回false，因此进入while循环中，并调用LockSupport中的park方法将当前线程进行阻塞。

   同步队列和等待队列似乎是同一个队列？只是通过Node节点的等待状态位waitStatus来进行区分吗？

public final void await() throws InterruptedException {
    // 对被中断的线程调用await()方法, 产生异常
    if (Thread.interrupted()){
        throw new InterruptedException();        
    }

    // 将当前线程添加到等待队列中
    Node node = addConditionWaiter();

    // 释放当前线程添加的全部的可重入锁
    int savedState = fullyRelease(node);

    int interruptMode = 0;

    while (!isOnSyncQueue(node)) {
        // 调用LockSupport中的park()方法阻塞当前线程, 具体来说, this是Condition对象
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }

    // acquireQueued()中包含获取锁的操作
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

final boolean isOnSyncQueue(Node node) {
    // 状态为等待状态, 或者是等待队列的第一个, 说明不在同步队列中. 这是为什么?
    // 首次调用时, 二者均满足, 直接返回false
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;

    // 调用signalAll之后, 等待队列中的节点的waitStatus值会变成(第1个是自动生成的head节点)-1,-1,...,-1,0(最后一个是0). 被唤醒的线程只会从第2个节点开始, 因此上面判断如果node.prev == null, 说明是添加到等待队列中. 
    // 总结: 上面是添加到等待队列, 下面是从等待队列中唤醒
    if (node.next != null) // If has successor, it must be on queue
        return true;
        /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
    // 从同步队列的尾部向头部开始遍历来唤醒
    // 这里不可能是添加到等待队列中, 因为waitStatus==-2时, 就会走第一个if判断条件返回
    // 在正常情况下, 这里的作用仅仅判断了tail这个在等待队列中, 且tail.next == null的需要被唤醒的节点, 因为tail节点无法从上面的if中返回
    return findNodeFromTail(node);
}


final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 对于先进先出的公平锁, 那第一个被唤醒的线程对应第一个等待队列中的第一个node, 此时p = node.prev, 即p为head
            // TODO: 但是对于非公平锁, p == head便有可能不成立吗? 好像也成立?
            // tryAcquire(arg) 尝试获取锁
            if (p == head && tryAcquire(arg)) {
                // head节点向后移动
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}


// 获取非公平锁
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    // state值为0才可以被线程去获取, 否则说明已经有线程占用了锁
    int c = getState();
    if (c == 0) {
        // 通过CAS操作来尝试添加锁
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        // 如果当前线程是锁的拥有者, 获取可重入锁
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    // 除了上诉两种情况外, 不可以获取锁
    return false;
}

signalAll() 方法

执行流程：

1. 判断调用signalAll()方法的线程是否是锁的拥有者（鉴权）
2. 调用doSignalAll()方法唤醒等待队列中的线程
   1. doSignalAll() 方法会清空等待队列，同时将链表中的节点一个个拆散
   2. 对等待队列中的每一个线程调用transferForSignal()方法
      1. 使用CAS操作，将线程的等待状态位由-2（CONDITION)设置为0。如果没有修改成功，说明signal()操作取消

public final void signalAll() {
    // 判断调用signalAll的线程是否持有锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();

    Node first = firstWaiter;
    if (first != null)
        doSignalAll(first);
}

private void doSignalAll(Node first) {
    // 等待队列靠firstWaiter和lastWaiter这两个指针, 这里相当于把等待队列清空
    lastWaiter = firstWaiter = null;

    // 等待队列从头到尾, 通过nextWaiter指针, 并不是next指针, 依次调用transferForSignal()方法
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}


final boolean transferForSignal(Node node) {
    /*
         * If cannot change waitStatus, the node has been cancelled.
         */
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    /*
         * Splice onto queue and try to set waitStatus of predecessor to
         * indicate that thread is (probably) waiting. If cancelled or
         * attempt to set waitStatus fails, wake up to resync (in which
         * case the waitStatus can be transiently and harmlessly wrong).
         */

    // 返回node节点的前驱节点, 对于等待队列的链表头, 返回的是最初的那个tail, 目前tail已经指向最新入队的Node节点
    Node p = enq(node);
    int ws = p.waitStatus;
    // 如果前驱节点的等待状态位大于0 或者 通过CAS操作无法将p对应的线程的ws设置为SIGNAL, 那么唤醒当前线程node.thread
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)){
        // 唤醒node节点对应的线程
        // 目前来看, 这里应该就是关键, 对应await()方法产生的LockSupport.park()
        LockSupport.unpark(node.thread);        
    }
    return true;
}

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 在等待队列中的Node的prev和next均为null, 在这里才为Node的prev和next进行设置
            node.prev = t;
            // 设置不仅建立t和node的关系, 还将tail设置为新插入的node
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

// 判断Condition对象是否由当前对象持有, 即拥有锁才能调用Condition中的signal()和await()方法
protected final boolean isHeldExclusively() {
    // While we must in general read state before owner,
    // we don't need to do so to check if current thread is owner
    return getExclusiveOwnerThread() == Thread.currentThread();
}