Java高并发编程基础之AQS

引言

曾经有一道比较比较经典的面试题“你可以说说java的并发包下面有哪些常见的类？”大多数人应该均可以说出
CountDownLatch、CyclicBarrier、Sempahore多线程并发三大利器。这三大利器都是经过AbstractQueuedSynchronizer抽象类(下面简写AQS)来实现的，因此学习三大利器以前咱们有必要先来学习下AQS。

AQS是一种提供了原子式管理同步状态、阻塞和唤醒线程功能以及队列模型的简单框架

AQS结构

说到同步咱们如何来保证同步？你们第一印象确定是加锁了，说到锁的话你们确定首先会想到的是Synchronized。
Synchronized你们应该基本上都会使用，加锁和释放锁都是jvm 来帮咱们实现的，咱们只须要简单的加个 Synchronized关键字就能够了。
用起来超级方便。可是有没有一种状况咱们设置一个锁的超时时间Synchronized就有点实现不了，这时候咱们就能够用ReentrantLock来实现,ReentrantLock是经过aqs来实现的，今天咱们就经过ReentrantLock来学习一下aqs。

CAS && 公平锁和非公平锁

AQS里面用到了大量的CAS学习AQS以前咱们仍是有必要简单的先了解下CAS、公平锁和非公平锁。

CAS

• CAS 全称是 compare and swap，是一种用于在多线程环境下实现同步功能的机制。CAS 操做包含三个操做数 -- 内存位置、预期数值和新值。CAS 的实现逻辑是将内存位置处的数值与预期数值想比较，若相等，则将内存位置处的值替换为新值。若不相等，则不作任何操做,这个操做是个原子性操做，java里面的AtomicInteger等类都是经过cas来实现的。

  公平锁和非公平锁

• 公平锁：多个线程按照申请锁的顺序去得到锁，线程会直接进入队列去排队，队列中第一个才能得到到锁。
  优势：等待锁的线程不会饿死，每一个线程均可以获取到锁。
  缺点：总体吞吐效率相对非公平锁要低，等待队列中除第一个线程之外的全部线程都会阻塞，CPU唤醒阻塞线程的开销比非公平锁大。
• 非公平锁：多个线程去获取锁的时候，会直接去尝试获取，获取不到，再去进入等待队列，若是能获取到，就直接获取到锁。
  优势：能够减小CPU唤醒线程的开销，总体的吞吐效率会高点，CPU也没必要取唤醒全部线程，会减小唤起线程的数量。
  缺点：处于等待队列中的线程可能会饿死，或者等好久才会得到锁。
  文字有点拗口，咱们来个实际的例子说明下。好比咱们去食堂就餐的时候都要排队，你们都按照先来后到的顺序排队打饭，这就是公平锁。若是等到你准备拿盘子打饭的时候
  直接蹦出了一个五大三粗的胖子插队到你前面，你看打不赢他只能忍气吞声让他插队，等胖子打完饭了又来个小个子也来插你队，这时候你无法忍了，直接大吼一声让他滚，这个
  小个子只能屁颠屁颠到队尾去排队了这就是非公平锁。
  咱们先来看看AQS有哪些属性

  // 头结点
  private transient volatile Node head;

// 阻塞的尾节点，每一个新的节点进来，都插入到最后，也就造成了一个链表
private transient volatile Node tail;

// 这个是最重要的，表明当前锁的状态，0表明没有被占用，大于 0 表明有线程持有当前锁
// 这个值能够大于 1，是由于锁能够重入，每次重入都加上 1
private volatile int state;

// 表明当前持有独占锁的线程，举个最重要的使用例子，由于锁能够重入
// reentrantLock.lock()能够嵌套调用屡次，因此每次用这个来判断当前线程是否已经拥有了锁
// if (currentThread == getExclusiveOwnerThread()) {state++}
private transient Thread exclusiveOwnerThread; //继承自AbstractOwnableSynchronizer

下面咱们来写一个demo分析下lock 加锁和释放锁的过程
```java
   final void lock() {
            // 上来先试试直接把状态置位1，若是此时没人获取锁就直接
            if (compareAndSetState(0, 1))
                 // 争抢成功则修改得到锁状态的线程
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

cas尝试失败，说明已经有人再持有锁，因此进入acquire方法

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

tryAcquire方法，看名字大概能猜出什么意思，就是试一试。
tryAcquire其实是调用了父类Sync的nonfairTryAcquire方法

final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
             // 获取下当前锁的状态
            int c = getState();
            // 这个if 逻辑跟前面一进来就获取锁的逻辑同样都是经过cas尝试获取下锁
            if (c == 0) {
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 进入这个判断说明 锁重入了 状态须要进行+1
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                 // 若是锁的重入次数大于int的最大值，直接就抛出异常了，正常状况应该不存在这种状况，不过jdk仍是严谨的
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            // 返回false 说明尝试获取锁失败了，失败了就要进行acquireQueued方法了
            return false;
        }

tryAcquire方法若是获取锁失败了,那么确定就要排队等待获取锁。排队的线程须要待在哪里等待获取锁？这个就跟咱们线程池执行任务同样，线程池把任务都封装成一个work，而后当线程处理任务不过来的时候，就把任务放到队列里面。AQS一样也是相似的，把排队等待获取锁的线程封装成一个NODE。而后再把NODE放入到一个队列里面。队列以下所示，不过须要注意一点head是不存NODE的。

接下来咱们继续分析源码，看下获取锁失败是如何被加入队列的。
就要执行acquireQueued方法，执行acquireQueued方法以前须要先执行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;
            // cas 加入队列队尾
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        // 尾结点不为空 || cas 加入尾结点失败
        enq(node);
        return node;
    }

enq

接下来再看看enq方法

// 经过自旋和CAS必定要当前node加入队尾
private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            // 尾结点为空说明队列仍是空的，尚未被初始化，因此初始化头结点，能够看到头结点的node 是没有绑定线程的也就是不存数据的
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

经过addWaiter方法已经把获取锁的线程经过封装成一个NODE加入对列。上述方法的一个执行流程图以下：

，接下来就是继续执行acquireQueued方法

acquireQueued

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                 // 经过自旋去获取锁 前驱节点==head的时候去尝试获取锁，这个方法在前面已经分析过了。
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
               // 进入这个if说明node的前驱节点不等于head 或者尝试获取锁失败了
               // 判断是否须要挂起当前线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
               // 异常状况进入cancelAcquire，在jdk11的时候这个源码直接是catch (Throwable e){ cancelAcquire(node);} 简单明了
            if (failed)
                cancelAcquire(node);
        }
    }

setHead

这个方法每当有一个node获取到锁了，就把当前node节点设置为头节点，能够简单的看作当前节点获取到锁了就把当前节点”移除“（变为头结点）队列。

shouldParkAfterFailedAcquire

说到这个方法咱们就要先看下NODE可能会有哪些状态在源码里面咱们能够看到总共会有四种状态

• CANCELLED：值为1，在同步队列中等待的线程等待超时或被中断，须要从同步队列中取消该Node的结点，其结点的waitStatus为CANCELLED，即结束状态，进入该状态后的结点将不会再变化。
• SIGNAL：值为-1，被标识为该等待唤醒状态的后继结点，当其前继结点的线程释放了同步锁或被取消，将会通知该后继结点的线程执行。说白了，就是处于唤醒状态，只要前继结点释放锁，就会通知标识为SIGNAL状态的后继结点的线程执行。
• CONDITION：值为-2，与Condition相关，该标识的结点处于等待队列中，结点的线程等待在Condition上，当其余线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
• PROPAGATE：值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        // 前驱节点状态 若是这个状态为-1 则返回true，把当前线程挂起
        if (ws == Node.SIGNAL)
            return true;
        // 大于0，说明状态为CANCELLED 
        if (ws > 0) {
            do {
               // 删除被取消的node（让被取消的node成为一个没有引用的node等着下次GC被回收）
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 进入这里只能是 0，-2，-3。NODE节点初始化的时候waitStatus默认值是0，因此只有这里才有修改waitStatus的地方
            // 经过cas 把前驱节点的状态设置为-1，而后返回false ，外面调用这个方法的是个循环，又会调用一次这个方法
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

parkAndCheckInterrupt

挂起当前线程，而且阻塞

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this); // 挂起当前线程，阻塞
    return Thread.interrupted();
}

解锁

加锁成功了，那锁用完了就应该释放锁了，释放锁重点看下unparkSuccessor这个方法就行了

private void unparkSuccessor(Node node) {
          // 头结点状态
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        Node s = node.next;
        // s==null head的successor节点获取锁成功后，执行了head.next=null的操做后，解锁线程读取了head.next，所以s==null
        // head的successor节点被取消（cancelAcquire）时，执行了以下操做：successor.waitStatus=1 ; successor.next = successor;
        if (s == null || s.waitStatus > 0) {
            s = null;
            // 从尾节点开始往前找，找到最前面的非取消的节点 这里没有break 哦
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
             // 唤醒线程 ，唤醒的线程会从acquireQueued去获取锁
            LockSupport.unpark(s.thread);
    }

释放锁代码比较简单，基本都写在代码注释里面了，流程以下：

这段代码里面有一个比较经典的面试题：
若是头结点的下一个节点为空或者头结点的下一个节点的状态为取消的时候为何要从后往前找，找到最前面非取消的节点？

• node.prev = pred; compareAndSetTail(pred, node) 这两个地方能够看做Tail入队的原子操做，可是此时pred.next = node;还没执行，若是这个时候执行了unparkSuccessor方法，就没办法从前日后找了，因此须要从后往前找。
• 在产生CANCELLED状态节点的时候，先断开的是Next指针，Prev指针并未断开，所以也是必需要从后往前遍历才可以遍历彻底部的Node

  总结

• reentrantLock的获取锁和释放锁基本就讲完了，里面还涉及多比较多的细节，感兴趣的同窗能够对着源码一行一行去debug试试。
• 适当的了解aqs才能更好的学习CountDownLatch、CyclicBarrier、Sempahore，由于这三个利器都是基于aqs来实现的。

  结束

• 因为本身才疏学浅，不免会有纰漏，假如你发现了错误的地方，还望留言给我指出来,我会对其加以修正。
• 若是你以为文章还不错，你的转发、分享、赞扬、点赞、留言就是对我最大的鼓励。
• 感谢您的阅读,十分欢迎并感谢您的关注。
  
  站在巨人的肩膀上摘苹果:
  https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html
  https://javadoop.com/post/AbstractQueuedSynchronizer
  https://www.cnblogs.com/yanlong300/p/10953185.html