Java 并发之AbstractQueuedSynchronizer(AQS)源码解析

// 共享模式
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;
    
// 最重要的信息，就是为上面的1，-1，-2，-3，以及0
volatile int waitStatus;
// 前置节点
volatile Node prev;
// 后置节点
volatile Node next;
// 节点存储的线程信息，会被挂起和唤醒
volatile Thread thread;

// 和condition 有关
Node nextWaiter;

// CLH队列的头结点，其不包含线程信息，head永远为null
private transient volatile Node head;
// CLH队列的尾节点，每次新加一个节点都会尾插到最后
private transient volatile Node tail;
// 当前锁被占据的次数，因为可以被一个线程重复占据，所以其值可以大于0
// 没有线程占据，其值就是0
private volatile int state;
// 当前运行的线程，也是占据锁的线程，注意和CLH中的线程无关
private transient Thread exclusiveOwnerThread; 

Lock lock = new ReentrantLock();
// 可以通过传入的boolean值设置其为公平方式还是不公平方式占据锁
lock.lock();
try { 
  // 业务代码，线程互斥
}
finally {
  lock.unlock(); 
}

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    // 获取当前线程，此处肯定是A线程了
    int c = getState();
    if (c == 0) {
        // 第一次运行，state=0
        if (compareAndSetState(0, acquires)) {
            // 1、利用CAS自旋方式，判断当前state确实为0，然后设置成acquire（1）
            // 这是原子性的操作，可以保证线程安全
            setExclusiveOwnerThread(current);
            // 设置当前执行的线程，直接返回为true
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
       // 2、当前的线程和执行中的线程是同一个，也就意味着可重入操作
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        // 表示当前锁被1个线程重复获取了nextc次
        return true;
    }
    // 否则就是返回false，表示没有尝试成功获取当前锁
    return false;
}

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // 新建一个节点，包装的线程是当前线程，传入的mode是互斥的
    // 当前线程是B线程
    Node pred = tail;
    // 此时尾节点tail=null
    if (pred != null) {
        // 当尾节点不为null，也就是CLH存在节点数据时
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            // 又是CAS操作，判断当前pred节点是否是尾节点，如果是，设置node为尾节点，否则跳过
            pred.next = node;
            // 双向链表的基本操作，把当前节点当做尾部节点插入
            return node;
        }
    }
    // 其实上面这一段代码的重点是**快速**，假定当前不存在互斥或者冲突，能够高效的添加节点成功
    
    enq(node);
    // 这个时候B线程创建好一个节点，添加到尾部
    return node;
    // 然后返回新创建的节点
}

private Node enq(final Node node) {
    for (;;) {
        // 死循环，可用严格确保可用完整插入到队列尾部
        Node t = tail;
        if (t == null) { 
            // 1、如果当前尾部是null，这个地方就是初始化的CLH队列的地方
            // B线程肯定会执行到这个地方
            if (compareAndSetHead(new Node()))
                // CAS自旋方式，设置当前队列头部节点为一个new Node（不包含线程信息）
                tail = head;
                // 头部和尾部指向同一个新建的new Node 节点
                // 注意此时没有退出。。。。只是完成if操作，会继续for循环操作
        } else {
            // 2、第二次循环到这里，CLH队列只有一个节点new Node
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                // 利用CAS的方式，配合双写链表的连接方式，添加一个节点到尾部
                t.next = node;
                // 然后返回原尾部节点
                return t;
            }
        }
    }
}

final boolean acquireQueued(final Node node, int arg) {
    // 这个时候node节点是上图中的tail节点，也就是线程B所在的节点，arg=1
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 查看当前node节点的前置节点p
            if (p == head && tryAcquire(arg)) {
               // 如果前置节点是头部节点，B线程进行尝试获取锁
               // 如果!是如果!
               // 尝试获取成功了，也就是A线程执行完成，然后设置当前当前可执行的节点是B线程
                setHead(node);
                // 把node节点的线程设置为null，然后head指到node上
                p.next = null; // help GC
                // 回收操作
                failed = false;
                // 返回，然后将执行的将会是B线程
                return interrupted;
            }
            // 按照我们的线程ABC的样例，上面的if肯定不会执行
            // 1、因为即使node的前节点是head，但是也无法获取到锁，因为一直被A线程占有者
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                // 如果shouldParkAfterFailedAcquire返回true，则挂起，否则不挂起线程
                interrupted = true;
            // 2、第一次循环设置了p节点状态位-1，然后B线程循环至这里被挂起
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

// 返回true，挂起，返回false则不挂起
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    
    if (ws == Node.SIGNAL)
        // 前节点的状态=-1，返回true，直接挂起即可
        return true;
    if (ws > 0) {
        // 前置节点状态有>0，说明前驱节点取消了排队，移除掉该节点
        do {
            node.prev = pred = pred.prev;
            // 拆分为2步
            // pred = pred.prev
            // node.prev = pred
        } while (pred.waitStatus > 0);
        // 当前节点开始往前扫描，从双向队列中除去>0的节点，直到<=0
        pred.next = node;
    } else {
        // 利用CAS设置前节点状态位-1
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
// 总结一下该方法用途：当前节点的前驱节点状态位-1时，挂起当前线程
// 返回false时，也就是不需要挂起，返回true，则需要调用parkAndCheckInterrupt挂起当前线程

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        // 当前运行的线程和锁住的线程不是同一个，抛出监控状态异常错误
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        // 如果为0，则意味着这时可以释放锁了
        free = true;
        setExclusiveOwnerThread(null);
        // 设置当前的占据锁的线程为null，给其他线程一个机会
    }
    setState(c);
    // 存在重入的机会，所以不一定就是释放该锁
    return free;
}

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        // 返回true，则表示可以唤醒其他挂起的线程，通知他们可以干活了
        Node h = head;
        if (h != null && h.waitStatus != 0)
            // 如果头部节点有效，则开始唤醒CLH队列的节点线程
            unparkSuccessor(h);
        return true;
    }
    return false;
}

private void unparkSuccessor(Node node) {
    // node节点是head，在我们上面已经说明了head节点的waitStatus=-1
   
    int ws = node.waitStatus;
    if (ws < 0)
        // 1、设置head节点状态为0
        compareAndSetWaitStatus(node, ws, 0);

    Node s = node.next;
    // 从head下一个节点开始处理
    if (s == null || s.waitStatus > 0) {
        s = null;
        // 如果下一个节点为null，或者下一个节点状态>0（为废弃节点）
        // 从尾节点开始扫描，直到扫描到距离head最近的一个<=0的节点信息
        // 在我们当前的例子中，肯定就是线程B
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        // 样例中，s节点就是B线程所在的节点
        // 2、唤醒s.thread ，会触发acquireQueued的循环继续执行
        LockSupport.unpark(s.thread);
}

for (;;) {
            final Node p = node.predecessor();
            // 查看当前node节点的前置节点p
            if (p == head && tryAcquire(arg)) {
               // 如果前置节点是头部节点，B线程进行尝试获取锁
               // 如果!是如果!
               // 尝试获取成功了，也就是A线程执行完成，然后设置当前当前可执行的节点是B线程
                setHead(node);
                // 把node节点的线程设置为null，然后head指到node上
                p.next = null; // help GC
                // 回收操作
                failed = false;
                // 返回，然后将执行的将会是B线程
                return interrupted;
            }

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

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
    // 利用另外的线程对当前线程触发中断请求，会直接返回当前中断状态，也就是true
    // 那么在acquireQueued 函数中 interrupted 的值被赋值为true
    // 其他有任何的改变...
}

public final void acquireSharedInterruptibly(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())
                // 重点在这里，一旦出现了异常，则立马抛出该异常情况
                // 此时failed 字段肯定是true了
                // 需要进入到下面的cancelAcquire 方法中去取消竞争了
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            // 开始取消竞争
            cancelAcquire(node);
    }
}

private void cancelAcquire(Node node) {
    // Ignore if node doesn‘t exist
    if (node == null)
        return;

    node.thread = null;
    // 节点线程数据设置为null

    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;
    // 从当前节点的前置节点开始扫描去掉废弃的节点，这样操作不会出现并发问题

    Node predNext = pred.next;

    // 设置为取消状态！
    node.waitStatus = Node.CANCELLED;

    // If we are the tail, remove ourselves.
    if (node == tail && compareAndSetTail(node, pred)) {
        // node是尾部节点，而且通过CAS自旋成功，设置pred为tail
        // pred的next为nul，可以彻底和node 节点切割，node节点等待GC回收
        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;
        if (pred != head &&
            // 前节点不是头部节点，而且包含的线程不为null（上面就有操作设置线程为null的情况）
            // 或者节点状态可以设置为-1
            ((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);
                // CAS 操作，设置后置节点
             // 这个操作就相当去双写链表去掉node节点   
        } else {
            unparkSuccessor(node);
            // 唤醒node的后置节点
            // 注意！此时node节点还在链表中
            // 他的真正移除是依靠node后置节点的shouldParkAfterFailedAcquire 去移除节点操作完成的
        }
        node.next = node; // help GC
    }
}

Node pred = tail;
if (pred != null) {
    node.prev = pred;
    if (compareAndSetTail(pred, node)) {
        // 通过CAS 操作后node节点为tail
        // 并且早就设置了node.prev = pred
        // 1、也就是前置的链表已经绑定完毕，后置的链表还未绑定
        pred.next = node;
        return node;
    }
}

public final boolean hasQueuedPredecessors() {
    Node t = tail; // Read fields in reverse initialization order
    Node h = head;
    // 先读取tail节点，再读取head节点
    Node s;
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}