【并发编程篇】从源码层面深入理解AQS

AQS是JUC提供的一个用于构建锁和同步容器的基础类。JUC包内许多类都是基于AQS构建的，例如ReentrantLock、Semaphore、CountDownLatch、ReentrantReadWriteLock、FutureTask等。AQS解决了在实现同步容器时设计的大量细节问题。
AQS是CLH队列的一个变种，主要原理和CLH队列差不多，这也是前面对CLH队列进行长篇大论介绍的原因。AQS队列内部维护的是一个FIFO的双向链表，这种结构的特点是每个数据结构都有两个指针，分别指向直接的前驱节点和直接的后继节点。所以双向链表可以从任意一个节点开始很方便地访问前驱节点和后继节点。每个节点其实是由线程封装的，当线程争抢锁失败后会封装成节点加入AQS队列中；当获取锁的线程释放锁以后，会从队列中唤醒一个阻塞的节点（线程）

//同步状态，使用 volatile保证线程可见
private volatile int state;

// 获取同步的状态
protected final int getState() {
    return state;
}

// 设置同步的状态
protected final void setState(int newState) {
    state = newState;
}
// 通过CAS设置同步的状态
protected final boolean compareAndSetState(int expect, int update) {
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

 static final class Node {
     
         /**节点等待状态值1：取消状态*/
         static final int CANCELLED =  1;
     
         /**节点等待状态值-1：标识后继线程处于等待状态*/
         static final int SIGNAL    = -1;
     
         /**节点等待状态值-2：标识当前线程正在进行条件等待*/
         static final int CONDITION = -2;
     
         /**节点等待状态值-3：标识下一次共享锁的acquireShared操作需要无条件传播*/
         static final int PROPAGATE = -3;
      
         //节点状态：值为SIGNAL、CANCELLED、CONDITION、PROPAGATE、0 
         //普通的同步节点的初始值为0，条件等待节点的初始值为CONDITION（-2）
         volatile int waitStatus;
     
         //节点所对应的线程，为抢锁线程或者条件等待线程
         volatile Thread thread;
     
         //前驱节点，当前节点会在前驱节点上自旋，循环检查前驱节点的waitStatus状态
         volatile Node prev;
     
         //后继节点
         volatile Node next;

         //若当前Node不是普通节点而是条件等待节点，则节点处于某个条件的等待队列上
         //此属性指向下一个条件等待节点，即其条件队列上的后继节点
         Node nextWaiter;
          ...
     }

static final class Node {
    //标识节点在抢占共享锁
    static final Node SHARED = new Node(); 
    //标识节点在抢占独占锁
    static final Node EXCLUSIVE = null;
    ...
}

/*首节点的引用*/
private transient volatile Node head;
/*尾节点的引用*/
private transient volatile Node tail;

@Slf4j
public class ConditionTest {

    public static void main(String[] args) {

        Lock lock = new ReentrantLock();
        Condition condition = lock.newCondition();

        new Thread(() -> {
            lock.lock();
            try {
                log.debug(Thread.currentThread().getName() + " 开始处理任务");
                condition.await();
                log.debug(Thread.currentThread().getName() + " 结束处理任务");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }).start();

        new Thread(() -> {
            lock.lock();
            try {
                log.debug(Thread.currentThread().getName() + " 开始处理任务");

                Thread.sleep(2000);
                condition.signal();
                log.debug(Thread.currentThread().getName() + " 结束处理任务");
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }).start();
    }
}

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

protected boolean tryRelease(long arg) {
    throw new UnsupportedOperationException();
}

protected long tryAcquireShared(long arg) {
    throw new UnsupportedOperationException();
}

protected boolean tryReleaseShared(long arg) {
    throw new UnsupportedOperationException();
}

protected boolean isHeldExclusively() {
    throw new UnsupportedOperationException();
}

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

public class DreamerLock extends AbstractQueuedSynchronizer{

    @Override
    protected boolean tryAcquire(int unused) {
        //cas 加锁  state=0
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    @Override
    protected boolean tryRelease(int unused) {
        //释放锁
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock() {
        acquire(1);
    }

    public boolean tryLock() {
        return tryAcquire(1);
    }

    public void unlock() {
        release(1);
    }

    public boolean isLocked() {
        return isHeldExclusively();
    }


}

private Node addWaiter(Node mode) {
    //创建新节点
    Node node = new Node(Thread.currentThread(), mode);
// 加入队列尾部，将目前的队列tail作为自己的前驱节点pred
Node pred = tail;
// 队列不为空的时候
if (pred != null) {
    node.prev = pred;
    // 先尝试通过AQS方式修改尾节点为最新的节点
    // 如果修改成功，将节点加入队列的尾部
    if (compareAndSetTail(pred, node)) {
        pred.next = node;
        return node;
    }
}
//第一次尝试添加尾部失败，意味着有并发抢锁发生，需要进行自旋
enq(node);
return node;
}

Node node = new Node(Thread.currentThread(), mode);

static final class Node {                 /** 常量标识：标识当前的队列节点类型为共享型抢占 */
    static final Node SHARED = new Node();           /** 常量标识：标识当前的队列节点类型为独占型抢占 */
    static final Node EXCLUSIVE = null;           // 省略其他代码      
}

/**
* 这里进行了循环，如果此时存在tail，就执行添加新队尾的操作
* 如果依然不存在，就把当前线程作为head节点
* 插入节点后，调用acquireQueued()进行阻塞
*/
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) {
            //队列为空，初始化尾节点和头节点为新节点
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 队列不为空，将新节点插入队列尾部
            node.prev = t; 
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

/**
* CAS 操作head指针，仅仅被enq()调用
*/
private final boolean compareAndSetHead(Node update) {
    return unsafe.compareAndSwapObject(this, headOffset, null, update);
}

/**
* CAS 操作tail指针，仅仅被enq()调用
*/
private final boolean compareAndSetTail(Node expect, Node update) {
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 自旋检查当前节点的前驱节点是否为头节点，才能获取锁
        for (;;) {
            // 获取节点的前驱节点
            final Node p = node.predecessor();
            // 节点中的线程循环地检查自己的前驱节点是否为 head 节点
            // 前驱节点是head时，进一步调用子类的tryAcquire（…）实现
            if (p == head && tryAcquire(arg)) {
                // tryAcquire()成功后，将当前节点设置为头节点，移除之前的头节点
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 检查前一个节点的状态，预判当前获取锁失败的线程是否要挂起
            // 如果需要挂起
            // 调用parkAndCheckInterrupt()方法挂起当前线程，直到被唤醒
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;                // 若两个操作都是true，则置true
        }

    } finally {
        //如果等待过程中没有成功获取资源（如timeout，或者可中断的情况下被中断了）
        //那么取消节点在队列中的等待
        if (failed)
            //取消请求，将当前节点从队列中移除
            cancelAcquire(node);
    }
}

private static boolean shouldParkAfterFailedAcquire(
     Node pred, Node node) {
             int ws = pred.waitStatus; // 获得前驱节点的状态
             if (ws == Node.SIGNAL)   //如果前驱节点状态为SIGNAL（值为-1）就直接返回
                    return true;
             if (ws > 0) {   // 前驱节点以及取消CANCELLED（1） 
               do {
                   // 不断地循环，找到有效前驱节点，即非CANCELLED（值为1）类型节点
     
                   // 将pred记录前驱的前驱
                   pred = pred.prev; 
                   // 调整当前节点的prev指针，保持为前驱的前驱
                  node.prev = pred;
     
                } while (pred.waitStatus > 0);
     
                 // 调整前驱节点的next指针
                 pred.next = node;
     
             } else {
                // 如果前驱状态不是CANCELLED，也不是SIGNAL，就设置为SIGNAL
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
                // 设置前驱状态之后，此方法返回值还是为false，表示线程不可用，被阻塞
             }
             return false;
         }

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);                        // 调用park()使线程进入waiting状态
    return Thread.interrupted();           // 如果被唤醒，查看自己是否已经被中断
}

  private Node enq(final Node node) {
             for (;;) { //自旋入队
                 Node t = tail;
                 if (t == null) {
                      //队列为空，初始化尾节点和头节点为新节点
                     if (compareAndSetHead(new Node()))
                         tail = head;
                 } else {
                     //如果队列不为空，将新节点插入队列尾部
                     node.prev = t; 
                     if (compareAndSetTail(t, node)) {
                         t.next = node;
                         return t;
                     }
                 }
             }
         }

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 在前驱节点上自旋
        for (;;) {
            // 获取节点的前驱节点
            final Node p = node.predecessor();
            // （1）前驱节点是头节点
            // （2）通过子类的tryAcquire()钩子实现抢占成功
            if (p == head && tryAcquire(arg)) {
                // 将当前节点设置为头节点，之前的头节点出队
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 省略park（无限期阻塞）线程的代码
        }
    } finally {
        // 省略其他 
    }
}

public final boolean release(long arg) {
    if (tryRelease(arg)) {                 //释放锁的钩子方法的实现
    Node h = head;                     //队列头节点
    if (h != null && h.waitStatus != 0)
        unparkSuccessor(h);    //唤醒后继线程
    return true;
}
return false;
}

//unparkSuccessor的核心代码如下：

private void unparkSuccessor(Node node) {
    // 省略不相关代码
    Node s = node.next;                    //后继节点
    // 省略不相关代码
    if (s != null)
        LockSupport.unpark(s.thread);  //唤醒后继节点的线程
}