AQS

juc.locks包下

AbstractQueuedSynchronizer，抽象的队列同步器

aqs是用来构建锁或者其它同步器组件的重量级基础框架及整个JUC体系的基石， 通过内置的FIFO队列来完成资源获取线程的排队工作，并通过一个int类变量表示持有锁的状态，0未占用，大于0被占用。

解释：

抢到资源的线程直接使用办理业务，抢占不到资源的线程的必然涉及一种排队等候机制，抢占资源失败的线程继续去等待(类似办理窗口都满了，暂时没有受理窗口的顾客只能去候客区排队等候)，仍然保留获取锁的可能且获取锁流程仍在继续(候客区的顾客也在等着叫号，轮到了再去受理窗口办理业务）。

既然说到了排队等候机制，那么就一定会有某种队列形成，这样的队列是什么数据结构呢?

如果共享资源被占用，就需要一定的阻塞等待唤醒机制来保证锁分配。这个机制主要用的是CLH队列的变体实现的，将暂时获取不到锁的线程加入到队列中，这个队列就是AQS的抽象表现。它将请求共享资源的线程封装成队列的结点(Node) ，通过CAS、自旋以及LockSuport.park()的方式，维护state变量的状态，使并发达到同步的效果。

AQS相关

Semaphore、CyclicBarrier、ReentrantReadWriteLock、CountDownLatch、ReentrantLock等juc类都是AQS实现的，都有sync内部类。

可以说，AQS是锁的实现者，是Java并发大神Douglee，提出统一规范并简化了锁的实现，屏蔽了同步状态管理、阻塞线程排队和通知、唤醒机制等。

AQS初步知识

AQS使用一个volatile的int类型的成员变量来表示同步状态，通过内置的 FIFO队列来完成资源获取的排队工作将每条要去抢占资源的线程封装成 一个Node节点来实现锁的分配，通过CAS完成对State值的修改。

state变量+CLH双端Node队列

int变量：

AQS的同步状态State成员变量

相当于银行办理业务的受理窗口状态，零就是没人，自由状态可以办理，大于等于1，有人占用窗口，等着去。

CLH队列队列：

CLH队列（三个大牛的名字组成），为一个双向队列，排队线程

内部类Node

Node类在AQS类内部，结构如下：

队列中每个排队的个体就是一个Node，Node的等待状态waitState成员变量，一个Node相当于一个Thread线程（有volatile Thread变量）。

属性说明：

同步队列的基本结构

AQS底层是用LockSupport.park()来进行排队的

ReentrantLock开始解读AQS

Lock接口的实现类，基本都是通过【聚合】了一个【队列同步器】Sync的子类完成线程访问控制的

公平锁和非公平锁

ReentrantLock底层实现：

可以明显看出公平锁与非公平锁的lock()方法唯一的区别就在于公平锁在获取同步状态时多了一个限制条件:hasQueuedPredecessors()。hasQueuedPredecessors是公平锁加锁时判断等待队列中是否存在有效节点的方法，是否排队。

解读非公平锁方法lock()中的AQS

底层AQS源码深度分析

    public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();

        //带入一个银行办理业务的案例来模拟我们的AQS如何进行线程的管理和通知唤醒机制

        //3个线程模拟3个来银行网点，受理窗口办理业务的顾客

        //A顾客就是第一个顾客，此时受理窗口没有任何人，A可以直接去办理
        new Thread(() -> {
            lock.lock();
            try {
                System.out.println("-----A thread come in");

                try {
                    TimeUnit.MINUTES.sleep(20);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } finally {
                lock.unlock();
            }
        }, "A").start();

        //第二个顾客，第二个线程---》由于受理业务的窗口只有一个(只能一个线程持有锁)，此时B只能等待，
        //进入候客区
        new Thread(() -> {
            lock.lock();
            try {
                System.out.println("-----B thread come in");
            } finally {
                lock.unlock();
            }
        }, "B").start();

        //第三个顾客，第三个线程---》由于受理业务的窗口只有一个(只能一个线程持有锁)，此时C只能等待，
        //进入候客区
        new Thread(() -> {
            lock.lock();
            try {
                System.out.println("-----C thread come in");
            } finally {
                lock.unlock();
            }
        }, "C").start();
    }

lock()方法：

state变量为0，则未锁住，其中一个线程A抢占cpu，置为1，CAS比较并交换(usafe类)

acquire()方法（AQS类中）：

cpu被抢占了，state > 0，后面线程B，C等待，准备入队列

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

一共有3个方法：

tryAcquire()

尝试获取锁，抢占。模板设计模式，子类不重写，就抛出异常，强制重写。

非公平锁的重写方法底层是nonfairTryAcquire方法：

    /**
      * Performs non-fair tryLock.  tryAcquire is implemented in
      * subclasses, but both need nonfair try for trylock method.
      */
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();

        int c = getState();  
        //刚好cpu空闲，线程B正好抢到
        if (c == 0) { 
            if (compareAndSetState(0, acquires)) {    
                setExclusiveOwnerThread(current);  
                return true;
            }         
        }

        //可重入锁，抢到cpu的线程跟当前线程是同一个则获取到锁，state+1
        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;  
    }

addWaiter(Node.EXCLUSIVE)

EXCLUSIVE排它，队列是双向队列

    /**
     * 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) {
        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) {
            //线程C指向尾结点B
            node.prev = pred;
            //设置线程C为尾结点
            if (compareAndSetTail(pred, node)) {
                //尾结点B,之前的尾结点B下一个结点为C结点
                pred.next = node;
                return node;
            }
        }
        
        //第一个到达队列的线程，执行这里，进队列
        enq(node);
        return node;
    }

双向链表中，第一个节点为虚节点(也叫哨兵节点)，其实并不存储任何信息，只是占位。 真正的第一个有数据的节点，是从第二个节点开始的。enq(node)添加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) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                //第一次进队列的头结点不是当前线程结点，是一个空节点，叫做哨兵结点
                if (compareAndSetHead(new Node()))
                    //尾结点也指向哨兵结点
                    tail = head;
            } else {
                //有了哨兵结点，执行，当前结点的前指针指向尾结点
                node.prev = t;
                //设置新进的线程结点为尾结点
                if (compareAndSetTail(t, node)) {
                    //上一个尾结点的next指针指向最新的尾结点，即是当前线程结点
                    t.next = node;
                    return t;
                }
            }
        }
    }

第一个线程B到达队列后执行如下：

第二个线程C入队列：

acquireQueued()

    /**
     * Acquires in exclusive uninterruptible mode for thread already in
     * queue. Used by condition wait methods as well as acquire.
     *
     * @param node the node
     * @param arg the acquire argument
     * @return {@code true} if interrupted while waiting
     */
    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            //线程中断标志
            boolean interrupted = false;
            //自旋
            for (;;) {
                //获取当前线程的上一个结点
                final Node p = node.predecessor();
                //p == head，说明是第一个线程结点（头结点是空节点），出列再次尝试抢占cpu
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                //尝试抢占cpu失败，LockSupport,park阻塞当前线程，等待被唤醒，唤醒后再次自旋tryAcquire去尝试获取cpu
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

park阻塞当前线程前再次尝试抢占cpu，只有前一结点状态是-1，才执行阻塞，方法shouldParkAfterFailedAcquire：

    /**
     * Checks and updates status for a node that failed to acquire.
     * Returns true if thread should block. This is the main signal
     * control in all acquire loops.  Requires that pred == node.prev.
     *
     * @param pred node's predecessor holding status
     * @param node the node
     * @return {@code true} if thread should block
     */
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        //当前结点的prev结点状态，起始0
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely 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 must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            //上一结点的结点状态设置为-1
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

只有上一结点的状态为SIGNAL -1时，才会执行parkAndCheckInterrupt()方法park阻塞当前线程：

   /**
     * Convenience method to park and then check if interrupted
     *
     * @return {@code true} if interrupted
     */
    private final boolean parkAndCheckInterrupt() {
        //卡死在这，等待被唤醒
        LockSupport.park(this);
        return Thread.interrupted();
    }

当占用cpu的线程A被释放时，线程B出队列，设置线程B结点的thread为null，变为哨兵结点，具体方法setHead()：

    /**
     * Sets head of queue to be node, thus dequeuing. Called only by
     * acquire methods.  Also nulls out unused fields for sake of GC
     * and to suppress unnecessary signals and traversals.
     *
     * @param node the node
     */
    private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    } 

原哨兵结点等待被GC，线程B结点变新的哨兵结点：

解读非公平锁方法unLock()中的AQS

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

    /**
     * Releases in exclusive mode.  Implemented by unblocking one or
     * more threads if {@link #tryRelease} returns true.
     * This method can be used to implement method {@link Lock#unlock}.
     *
     * @param arg the release argument.  This value is conveyed to
     *        {@link #tryRelease} but is otherwise uninterpreted and
     *        can represent anything you like.
     * @return the value returned from {@link #tryRelease}
     */
    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            //释放成功
            //哨兵结点
            Node h = head;
            if (h != null && h.waitStatus != 0)
                //确定是阻塞状态
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

tryRelease()方法是抛异常，模板设计模式方法，强制重写，ReentrantLock中内部类Sync重写。

    protected final boolean tryRelease(int releases) {
        //为0，则准备释放
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            //设置cpu占用线程为null，空闲，e s ku save bu
            setExclusiveOwnerThread(null);
        }
        //设置state=0,可重入锁一直减1
        setState(c);
        return free;
    }

唤醒线程结点方法unparkSuccessor()：

    /**
     * Wakes up node's successor, if one exists.
     *
     * @param node the node
     */
    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;
        //头结点第一次进来设置状态为0
        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);
    }

AQS面试题

我相信你应该看过源码了，那么AQS里面有个变量叫State，它的值有几种？

3个状态：没占用是0，占用了是1，大于1是可重入锁

如果AB两个线程进来了以后，请问这个总共有多少个Node节点？

答案是3个，null的哨兵头结点