JUC多并发编程 AQS

基础解释：

是用来实现锁或者其他同步器组件的公共基础部分的抽象实现,是重量级基础框架及整个JUC体系的基石，主要用于锁分配给“谁”的问题。
整体就是一个抽象的 FIFO 队列来完成资源获取线程的排队工作，并通过一个 int 类变量表示持有锁的状态

CLH: Craig、Landin and Hagersten 队列，是一个单向链表， AQS 中的队列是 CLH 变体的虚拟双向队列 FIFO

锁和同步器的关系：

锁，面向锁的使用者：定义了程序员和锁交互的使用层 API，隐藏了实现细节，调用即可
同步器，面向锁的实现者：提出统一规范并简化了锁的实现，将其抽象出来屏蔽了同步状态管理、同步队列的管理和维护、阻塞线程队列和通知、唤醒机制等，是一切锁和同步组件实现的公共基础部分

同步器的作用：

加锁会导致阻塞，有阻塞就需要排队，实现排队必然需要队列
抢到资源的线程直接使用处理业务，抢不到资源的必然涉及一种排队等候机制。抢占资源失败的线程继续去等待(类似银行业务办理窗口都满了，暂时没有受理窗口的顾客只能去候客区排队等待)，单等候线程仍然保留获取锁的可能且获取锁流程仍在继续(候客区的顾客也等着叫号,抢到了再去受理窗口办理业务)
如果共享资源被占用，就需要一定的阻塞等待唤醒机制来保证锁分配。这个机制主要用的是 CLH 队列的变体实现的，将暂时获取不到锁的线程加入到队列中，这个队列就是 AQS 同步队列抽象表现。它将要请求共享资源的线程及自身的等待状态封装成队列的节点对象(Node)，通过 CAS, 自旋以及 LockSupport.park()的方式, 维护 state 变量的状态，使并发达到同步的效果

AQS类图:

AQS 内部体系架构

AQS 自身:

AQS 的 int 变量，AQS 的同步状态 State 成员变量，0表示未占用，大于1表示有人占用
AQS 的 CLH 队列: 是一个双向队列，从尾部入队，从头部出对

Node 内部类:

Node 的等待状态 waitState 成员变量，表示等待状态

属性	说明
static final Node SHARED = new Node()	共享
static final Node EXCLUSIVE = null	独占
static final int CANCELLED = 1	线程被取消了
static final int SIGNAL = -1	后续线程需要唤醒
static final int CONDITION = -2	等待 condition 唤醒
static final int PROPAGATE = -3	共享式同步状态获取将会无条件得传播下去
volatile int waitStatus	初始为0，状态是上面的几种
volatile Node prev	前置节点
volatile Node next	后置节点

AQS 源码分析

ReentrantLock:

Lock 接口的实现类，基本都是通过聚合一个队列同步器的子类完成线程访问控制的
默认和fair为false情况下创建的是非公平锁, fair 为true 的情况下创建的是公平锁
对比公平锁和非公平锁的 tryAcquire() 方法的实现代码，其实差别就在于非公平锁获取锁时比公平锁中少了一个判断 !hasQueuedPredecessors()，该方法用来判断是否需要排队
公平锁：讲究先来先到，线程在获取锁时，如果这个锁的等待队列中已经有线程在等待，那么当前线程就会进入等待队列
非公平锁：不管是否等待队列，如果可以获取锁，则立刻占有锁对象，也就是说第一个队列线程苏醒后，不一定就是排头的这个线程获得锁，它还需要参加竞争锁

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    private final Sync sync;


    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;

        abstract void lock();
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                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;
        }

        protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }

        protected final boolean isHeldExclusively() {
            return getExclusiveOwnerThread() == Thread.currentThread();
        }

        final ConditionObject newCondition() {
            return new ConditionObject();
        }

        final Thread getOwner() {
            return getState() == 0 ? null : getExclusiveOwnerThread();
        }

        final int getHoldCount() {
            return isHeldExclusively() ? getState() : 0;
        }

        final boolean isLocked() {
            return getState() != 0;
        }

        private void readObject(java.io.ObjectInputStream s)
            throws java.io.IOException, ClassNotFoundException {
            s.defaultReadObject();
            setState(0); // reset to unlocked state
        }
    }


    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }


    static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;

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

        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

    public ReentrantLock() {
        sync = new NonfairSync();
    }

    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

    public void lock() {
        sync.lock();
    }

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

acquire:

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

    // AbstractQueuedSynchronizer 设计模式-模板，具体由子类实现
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }
    
    // ReentrantLock Sync 进行抢锁
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        // 资源空闲
        if (c == 0) {
            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;
   }

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

acquireQueued:

    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 static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 获取前驱节点状态
        int ws = pred.waitStatus;
        // 如果是 SIGNAL 状态，即等待被占用的资源释放, 直接返回 true
        // 准备继续调用 parkAndCheckInterrupt 方法
        if (ws == Node.SIGNAL)
            return true;
        // 说明是 CANCELLED 状态
        if (ws > 0) {
            do {
                // 循环判断前驱节点的前驱节点是否也为 CANCELLED 状态，忽略该状态节点，重新连接
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 将当前节点的前驱节点设置为 SIGNAL 状态，用于后续唤醒事件
            // 程序第一次执行到这里返回 false，会进入第二次循环
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }


    private final boolean parkAndCheckInterrupt() {
        // 线程挂起，程序不会继续向下执行
        // 解除会有三种情况，被 unpark, 被中断(interrupt), 其他不合逻辑的返回
        LockSupport.park(this);
        // 返回当前线程的中断状态，并清空中断状态
        // 如果由于被中断，会返回 true
        return Thread.interrupted();
    }

cancelAcquire:

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

unlock:

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

    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            // 不为空 并且 状态已经为 -1
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    // 模板
    protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }
    
    // 释放资源
    protected final boolean tryRelease(int releases) {
        int c = getState() - releases; // 0
        if (Thread.currentThread() != getExclusiveOwnerThread())
           throw new IllegalMonitorStateException();  // 基本不会出现
        boolean free = false;
        if (c == 0) {
           // 空闲
           free = true;
           setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }

unparkSuccessor:

    private void unparkSuccessor(Node node) {

        int ws = node.waitStatus;
        if (ws < 0)
            // 头节点状态设置为 0
            compareAndSetWaitStatus(node, ws, 0);

        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)
            // unpark 下一个节点
            LockSupport.unpark(s.thread);
    }