首页 Web开发 Windows程序 编程语言 数据库 移动开发 系统相关 微信 其他好文

java高并发编程基础之AQS

时间：2021-02-25 12:23:16   收藏：0   阅读：0

引言

曾经有一道比较比较经典的面试题“你能够说说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
公平锁和非公平锁
// 头结点
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 加锁和释放锁的过程

   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可能会有哪些状态在源码里面我们可以看到总共会有四种状态

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

释放锁代码比较简单，基本都写在代码注释里面了，流程如下：
技术图片
这段代码里面有一个比较经典的面试题：
如果头结点的下一个节点为空或者头结点的下一个节点的状态为取消的时候为什么要从后往前找，找到最前面非取消的节点？

总结

结束

评论（0）
© 2014 mamicode.com 版权所有 京ICP备13008772号-2  联系我们:gaon5@hotmail.com
迷上了代码！