深刻浅出AQS之独占锁模式

时间 2020-11-25

标签 java node 编程并发框架高并发性能学习 this 栏目 Java 繁體版

原文原文链接

每个Java工程师应该都或多或少了解过AQS，我本身也是前先后后，反反复复研究了好久，看了忘，忘了再看，每次都有不同的体会。此次趁着写博客，打算从新拿出来系统的研究下它的源码，总结成文章，便于之后复习。java

原文地址：http://www.jianshu.com/p/71449a7d01afnode

AbstractQueuedSynchronizer（如下简称AQS）做为java.util.concurrent包的基础，它提供了一套完整的同步编程框架，开发人员只须要实现其中几个简单的方法就能自由的使用诸如独占，共享，条件队列等多种同步模式。咱们经常使用的好比ReentrantLock，CountDownLatch等等基础类库都是基于AQS实现的，足以说明这套框架的强大之处。鉴于此，咱们开发人员更应该了解它的实现原理，这样才能在使用过程当中驾轻就熟。编程

整体来讲我的感受AQS的代码很是难懂，本文就其中的独占锁实现原理进行分析。并发

1、执行过程概述

首先先从总体流程入手，了解下AQS独占锁的执行逻辑，而后再一步一步深刻分析源码。框架

获取锁的过程：高并发

当线程调用acquire()申请获取锁资源，若是成功，则进入临界区。
当获取锁失败时，则进入一个FIFO等待队列，而后被挂起等待唤醒。
当队列中的等待线程被唤醒之后就从新尝试获取锁资源，若是成功则进入临界区，不然继续挂起等待。

释放锁过程：性能

当线程调用release()进行锁资源释放时，若是没有其余线程在等待锁资源，则释放完成。
若是队列中有其余等待锁资源的线程须要唤醒，则唤醒队列中的第一个等待节点（先入先出）。

2、源码深刻分析

基于上面所讲的独占锁获取释放的大体过程，咱们再来看下源码实现逻辑：
首先来看下获取锁的方法acquire()学习

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

代码虽然短，但包含的逻辑却不少，一步一步看下：ui

首先是调用开发人员本身实现的tryAcquire() 方法尝试获取锁资源，若是成功则整个acquire()方法执行完毕，即当前线程得到锁资源，能够进入临界区。
若是获取锁失败，则开始进入后面的逻辑，首先是addWaiter(Node.EXCLUSIVE)方法。来看下这个方法的源码实现：

//注意：该入队方法的返回值就是新建立的节点
    private Node addWaiter(Node mode) {
        //基于当前线程，节点类型（Node.EXCLUSIVE）建立新的节点
        //因为这里是独占模式，所以节点类型就是Node.EXCLUSIVE
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        //这里为了提搞性能，首先执行一次快速入队操做，即直接尝试将新节点加入队尾
        if (pred != null) {
            node.prev = pred;
            //这里根据CAS的逻辑，即便并发操做也只能有一个线程成功并返回，其他的都要执行后面的入队操做。即enq()方法
            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) { 
                //一样是CAS，只有一个线程能够初始化头结点成功，其他的都要重复执行循环体
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                //新建立的节点指向队列尾节点，毫无疑问并发状况下这里会有多个新建立的节点指向队列尾节点
                node.prev = t;
                //基于这一步的CAS，无论前一步有多少新节点都指向了尾节点，这一步只有一个能真正入队成功，其余的都必须从新执行循环体
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    //该循环体惟一退出的操做，就是入队成功（不然就要无限重试）
                    return t;
                }
            }
        }
    }

上面的入队操做有两点须要说明：
1、初始化队列的触发条件就是当前已经有线程占有了锁资源，所以上面建立的空的头节点能够认为就是当前占有锁资源的节点（虽然它并无设置任何属性）。
2、注意整个代码是处在一个死循环中，知道入队成功。若是失败了就会不断进行重试。this

通过上面的操做，咱们申请获取锁的线程已经成功加入了等待队列，经过文章最一开始说的独占锁获取流程，那么节点如今要作的就是挂起当前线程，等待被唤醒，这个逻辑是怎么实现的呢？来看下源码：

经过上面的分析，该方法入参node就是刚入队的包含当前线程信息的节点
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; //帮助GC
                    //表示锁资源成功获取，所以把failed置为false
                    failed = false;
                    //返回中断标记，表示当前节点是被正常唤醒仍是被中断唤醒
                    return interrupted;
                }
                若是没有获取锁成功，则进入挂起逻辑
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            //最后会分析获取锁失败处理逻辑
            if (failed)
                cancelAcquire(node);
        }
    }

挂起逻辑是很重要的逻辑，这里拿出来单独分析一下，首先要注意目前为止，咱们只是根据当前线程，节点类型建立了一个节点并加入队列中，其余属性都是默认值。

//首先说明一下参数，node是当前线程的节点，pred是它的前置节点
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        //获取前置节点的waitStatus
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            //若是前置节点的waitStatus是Node.SIGNAL则返回true，而后会执行parkAndCheckInterrupt()方法进行挂起
            return true;
        if (ws > 0) {
            //由waitStatus的几个取值能够判断这里表示前置节点被取消
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            //这里咱们由当前节点的前置节点开始，一直向前找最近的一个没有被取消的节点
            //注，因为头结点head是经过new Node()建立，它的waitStatus为0,所以这里不会出现空指针问题，也就是说最多就是找到头节点上面的循环就退出了
            pred.next = node;
        } else {
            //根据waitStatus的取值限定，这里waitStatus的值只能是0或者PROPAGATE，那么咱们把前置节点的waitStatus设为Node.SIGNAL而后从新进入该方法进行判断
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

上面这个方法逻辑比较复杂，它是用来判断当前节点是否能够被挂起，也就是唤醒条件是否已经具有，即若是挂起了，那必定是能够由其余线程来唤醒的。该方法若是返回false，即挂起条件没有完备，那就会从新执行acquireQueued方法的循环体，进行从新判断，若是返回true，那就表示万事俱备，能够挂起了，就会进入parkAndCheckInterrupt()方法看下源码：

private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        //被唤醒以后，返回中断标记，即若是是正常唤醒则返回false，若是是因为中断醒来，就返回true
        return Thread.interrupted();
    }

看acquireQueued方法中的源码，若是是由于中断醒来，那么就把中断标记置为true。不论是正常被唤醒仍是由与中断醒来，都会去尝试获取锁资源。若是成功则返回中断标记，不然继续挂起等待。
注：Thread.interrupted()方法在返回中断标记的同时会清除中断标记，也就是说当因为中断醒来而后获取锁成功，那么整个acquireQueued方法就会返回true表示是由于中断醒来，但若是中断醒来之后没有获取到锁，继续挂起，因为此次的中断已经被清除了，下次若是是被正常唤醒，那么acquireQueued方法就会返回false，表示没有中断。

最后咱们回到acquireQueued方法的最后一步，finally模块。这里是针对锁资源获取失败之后作的一些善后工做，翻看上面的代码，其实能进入这里的就是tryAcquire()方法抛出异常，也就是说AQS框架针对开发人员本身实现的获取锁操做若是抛出异常，也作了妥善的处理，一块儿来看下源码：

//传入的方法参数是当前获取锁资源失败的节点
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;

        //把当前节点waitStatus置为取消，这样别的节点在处理时就会跳过该节点
        node.waitStatus = Node.CANCELLED;
        //若是当前是尾节点，则直接删除，即出队
        //注：这里不用关心CAS失败，由于即便并发致使失败，该节点也已经被成功删除
        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)
                    //这里的判断逻辑很绕，具体就是若是当前节点的前置节点不是头节点且它后面的节点等待它唤醒（waitStatus小于0），
                    //再加上若是当前节点的后继节点没有被取消就把前置节点跟后置节点进行链接，至关于删除了当前节点
                    compareAndSetNext(pred, predNext, next);
            } else {
                //进入这里，要么当前节点的前置节点是头结点，要么前置节点的waitStatus是PROPAGATE，直接唤醒当前节点的后继节点
                unparkSuccessor(node);
            }

            node.next = node; // help GC
        }
    }

上面就是独占模式获取锁的核心源码，确实很是难懂，很绕，就这几个方法须要反反复复看不少遍，才能慢慢理解。

接下来看下释放锁的过程：

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()方法是用户自定义的释放锁逻辑，若是成功，就判断等待队列中有没有须要被唤醒的节点（waitStatus为0表示没有须要被唤醒的节点），一块儿看下唤醒操做：

private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
        if (ws < 0)
            //把标记为设置为0，表示唤醒操做已经开始进行，提升并发环境下性能
            compareAndSetWaitStatus(node, ws, 0);

        Node s = node.next;
        //若是当前节点的后继节点为null，或者已经被取消
        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)
            LockSupport.unpark(s.thread);
    }

相比而言，锁的释放操做就简单不少了，代码也比较少。

3、总结

以上就是AQS独占锁的获取与释放过程，大体思想很简单，就是尝试去获取锁，若是失败就加入一个队列中挂起。释放锁时，若是队列中有等待的线程就进行唤醒。但若是一步一步看源码，会发现细节很是多，不少地方很难搞明白，我本身也是反反复复学习好久才有点心得，但也不敢说已经研究通了AQS，甚至不敢说我上面的研究成果就是对的，只是写篇文章总结一下，跟同行交流交流心得。除了独占锁，后面还会产出AQS一系列的文章，包括共享锁，条件队列的实现原理等。