码迷，mamicode.com
首页 > 其他好文 > 详细

AQS

时间：2020-05-24 09:25:20      阅读：57      评论：0      收藏：0      [点我收藏+]

标签：data   自定义   name   不同   源代码   idt   try   href   else   

所谓AQS，指的是AbstractQueuedSynchronizer，它提供了一种实现阻塞锁和一系列依赖FIFO等待队列的同步器的框架，ReentrantLock、Semaphore、CountDownLatch、CyclicBarrier等并发类均是基于AQS来实现的，具体用法是通过继承AQS实现其模板方法，然后将子类作为同步组件的内部类。

了解一个框架最好的方式是读源码，说干就干。

AQS是JDK1.5之后才出现的，由大名鼎鼎的Doug Lea李大爷来操刀设计并开发实现，全部源代码(加注释)2315行，整体难度中等。

* @since 1.5
* @author Doug Lea

基本框架

在阅读源码前，首先阐述AQS的基本思想及其相关概念。

AQS基本框架如下图所示：

技术图片

 

AQS维护了一个volatile语义(支持多线程下的可见性)的共享资源变量state和一个FIFO线程等待队列(多线程竞争state被阻塞时会进入此队列)。

State

首先说一下共享资源变量state，它是int数据类型的，其访问方式有3种：

  • getState()
  • setState(int newState)
  • compareAndSetState(int expect, int update)

上述3种方式均是原子操作，其中compareAndSetState()的实现依赖于Unsafe的compareAndSwapInt()方法。

private volatile int state;

// 具有内存读可见性语义
protected final int getState() {
    return state;
}

// 具有内存写可见性语义
protected final void setState(int newState) {
    state = newState;
}

// 具有内存读/写可见性语义
protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

资源的共享方式分为2种：

  • 独占式(Exclusive)

只有单个线程能够成功获取资源并执行，如ReentrantLock。

  • 共享式(Shared)

多个线程可成功获取资源并执行，如Semaphore/CountDownLatch等。

AQS将大部分的同步逻辑均已经实现好，继承的自定义同步器只需要实现state的获取(acquire)和释放(release)的逻辑代码就可以，主要包括下面方法：

  • tryAcquire(int)：独占方式。尝试获取资源，成功则返回true，失败则返回false。
  • tryRelease(int)：独占方式。尝试释放资源，成功则返回true，失败则返回false。
  • tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
  • tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。
  • isHeldExclusively()：该线程是否正在独占资源。只有用到condition才需要去实现它。

AQS需要子类复写的方法均没有声明为abstract，目的是避免子类需要强制性覆写多个方法，因为一般自定义同步器要么是独占方法，要么是共享方法，只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。

当然，AQS也支持子类同时实现独占和共享两种模式，如ReentrantReadWriteLock。

CLH队列(FIFO)

AQS是通过内部类Node来实现FIFO队列的，源代码解析如下：

static final class Node {
    
    // 表明节点在共享模式下等待的标记
    static final Node SHARED = new Node();
    // 表明节点在独占模式下等待的标记
    static final Node EXCLUSIVE = null;

    // 表征等待线程已取消的
    static final int CANCELLED =  1;
    // 表征需要唤醒后续线程
    static final int SIGNAL    = -1;
    // 表征线程正在等待触发条件(condition)
    static final int CONDITION = -2;
    // 表征下一个acquireShared应无条件传播
    static final int PROPAGATE = -3;

    /**
     *   SIGNAL: 当前节点释放state或者取消后，将通知后续节点竞争state。
     *   CANCELLED: 线程因timeout和interrupt而放弃竞争state，当前节点将与state彻底拜拜
     *   CONDITION: 表征当前节点处于条件队列中，它将不能用作同步队列节点，直到其waitStatus被重置为0
     *   PROPAGATE: 表征下一个acquireShared应无条件传播
     *   0: None of the above
     */
    volatile int waitStatus;
    
    // 前继节点
    volatile Node prev;
    // 后继节点
    volatile Node next;
    // 持有的线程
    volatile Thread thread;
    // 链接下一个等待条件触发的节点
    Node nextWaiter;

    // 返回节点是否处于Shared状态下
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

    // 返回前继节点
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }
    
    // Shared模式下的Node构造函数
    Node() {  
    }

    // 用于addWaiter
    Node(Thread thread, Node mode) {  
        this.nextWaiter = mode;
        this.thread = thread;
    }
    
    // 用于Condition
    Node(Thread thread, int waitStatus) {
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

可以看到，waitStatus非负的时候，表征不可用，正数代表处于等待状态，所以waitStatus只需要检查其正负符号即可，不用太多关注特定值。

获取资源(独占模式)

acquire(int)

首先讲解独占模式(Exclusive)下的获取/释放资源过程，其入口方法为:

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

tryAcquire(arg)为线程获取资源的方法函数，在AQS中定义如下：

protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}

很明显，该方法是空方法，且由protected修饰，说明该方法需要由子类即自定义同步器来实现。

acquire()方法至少执行一次tryAcquire(arg)，若返回true，则acquire直接返回，否则进入acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法。

acquireQueued方法分为3个步骤：

  1. addWriter()将当前线程加入到等待队列的尾部，并标记为独占模式；
  2. acquireQueued()使线程在等待队列中获取资源，直到获取到资源返回，若整个等待过程被中断过，则返回True，否则返回False。
  3. 如果线程在等待过程中被中断过，则先标记上，待获取到资源后再进行自我中断selfInterrupt()，将中断响应掉。

下面具体看看过程中涉及到的各函数：

tryAcquire(int)

tryAcquire尝试以独占的模式获取资源，如果获取成功则返回True，否则直接返回False，默认实现是抛出UnsupportedOperationException，具体实现由自定义扩展了AQS的同步器来完成。

addWaiter(Node)

addWaiter为当前线程以指定模式创建节点，并将其添加到等待队列的尾部，其源码为：

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // 尝试将节点快速插入等待队列，若失败则执行常规插入(enq方法)
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 常规插入
    enq(node);
    return node;
}

再看enq(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;
            }
        }
    }
}

可以看到，常规插入与快速插入相比，有2点不同：

  1. 常规插入是自旋过程(for(;;))，能够保证节点插入成功；
  2. 比快速插入多包含了1种情况，即当前等待队列为空时，需要初始化队列，即将待插入节点设置为头结点，同时为尾节点(因为只有一个嘛)。

常规插入与快速插入均依赖于CAS，其实现依赖于unsafe类，具体代码如下：


private final boolean compareAndSetHead(Node update) {
    return unsafe.compareAndSwapObject(this, headOffset, null, update);
}

private final boolean compareAndSetTail(Node expect, Node update) {
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

unsafe中的cas操作均是native方法，由计算机CPU的cmpxchg指令来保证其原子性。

接着看acquireQueued()方法：

acquireQueued(Node, int)

相关说明已在代码中注释：

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);
                // 说明前继节点已经释放掉资源了，将其next置空，以方便虚拟机回收掉该前继节点
                p.next = null; // help GC
                // 标识获取资源成功
                failed = false;
                // 返回中断标记
                return interrupted;
            }
            // 若前继节点不是头结点，或者获取资源失败，
            // 则需要通过shouldParkAfterFailedAcquire函数
            // 判断是否需要阻塞该节点持有的线程
            // 若shouldParkAfterFailedAcquire函数返回true，
            // 则继续执行parkAndCheckInterrupt()函数，
            // 将该线程阻塞并检查是否可以被中断，若返回true，则将interrupted标志置于true
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        // 最终获取资源失败，则当前节点放弃获取资源
        if (failed)
            cancelAcquire(node);
    }
}

具体看一下shouldParkAfterFailedAcquire函数：

// shouldParkAfterFailedAcquire是通过前继节点的waitStatus值来判断是否阻塞当前节点的线程的
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取前继节点的waitStatus值ws
    int ws = pred.waitStatus;
    // 如果ws的值为Node.SIGNAL(-1)，则直接返回true
    // 说明前继节点完成资源的释放或者中断后，会通知当前节点的，回家等通知就好了，不用自旋频繁地来打听消息
    if (ws == Node.SIGNAL)
        return true;
    // 如果前继节点的ws值大于0,即为1,说明前继节点处于放弃状态(Cancelled)
    // 那就继续往前遍历，直到当前节点的前继节点的ws值为0或负数
    // 此处代码很关键，节点往前移动就是通过这里来实现的，直到节点的前继节点满足
    // if (p == head && tryAcquire(arg))条件，acquireQueued方法才能够跳出自旋过程
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 将前继节点的ws值设置为Node.SIGNAL，以保证下次自旋时，shouldParkAfterFailedAcquire直接返回true
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

parkAndCheckInterrupt()函数则简单很多，主要调用LockSupport类的park()方法阻塞当前线程，并返回线程是否被中断过。

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

至此，独占模式下，线程获取资源acquire的代码就跟完了，总结一下过程：

  1. 首先线程通过tryAcquire(arg)尝试获取共享资源，若获取成功则直接返回，若不成功，则将该线程以独占模式添加到等待队列尾部，tryAcquire(arg)由继承AQS的自定义同步器来具体实现；
  2. 当前线程加入等待队列后，会通过acquireQueued方法基于CAS自旋不断尝试获取资源，直至获取到资源；
  3. 若在自旋过程中，线程被中断过，acquireQueued方法会标记此次中断，并返回true。
  4. 若acquireQueued方法获取到资源后，返回true，则执行线程自我中断操作selfInterrupt()。
static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

释放资源(独占模式)

讲完获取资源，对应的讲一下AQS的释放资源过程，其入口函数为：

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        // 获取到等待队列的头结点h
        Node h = head;
        // 若头结点不为空且其ws值非0，则唤醒h的后继节点
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

逻辑并不复杂，通过tryRelease(arg)来释放资源，和tryAcquire类似，tryRelease也是有继承AQS的自定义同步器来具体实现。

tryRelease(int)

该方法尝试释放指定量的资源。

protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
}

unparkSuccessor(Node)

该方法主要用于唤醒等待队列中的下一个阻塞线程。

private void unparkSuccessor(Node node) {
    // 获取当前节点的ws值
    int ws = node.waitStatus;
    // 将当前节点的ws值置0
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    Node s = node.next;
    // 若后继节点为null或者其ws值大于0(放弃状态)，则从等待队列的尾节点从后往前搜索，
    // 搜索到等待队列中最靠前的ws值非正且非null的节点
    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;
    }
    // 如果后继节点非null，则唤醒该后继节点持有的线程
    if (s != null)
        LockSupport.unpark(s.thread);
}

后继节点的阻塞线程被唤醒后，就进入到acquireQueued()的if (p == head && tryAcquire(arg))的判断中，此时被唤醒的线程将尝试获取资源。

当然，如果被唤醒的线程所在节点的前继节点不是头结点，经过shouldParkAfterFailedAcquire的调整，也会移动到等待队列的前面，直到其前继节点为头结点。

讲解完独占模式下资源的acquire/release过程，下面开始讲解共享模式下，线程如何完成资源的获取和共享。

获取资源(共享模式)

理解了独占模式下，资源的获取和释放过程，则共享模式下也就so easy了，首先看一下方法入口：

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

执行tryAcquireShared方法获取资源，若获取成功则直接返回，若失败，则进入等待队列，执行自旋获取资源，具体由doAcquireShared方法来实现。

tryAcquireShared(int)

同样的，tryAcquireShared(int)由继承AQS的自定义同步器来具体实现。

protected int tryAcquireShared(int arg) {
    throw new UnsupportedOperationException();
}

其返回值为负值代表失败；0代表获取成功，但无剩余资源；正值代表获取成功且有剩余资源，其他线程可去获取。

doAcquireShared(int)

private void doAcquireShared(int arg) {
    // 将线程以共享模式添加到等待队列的尾部
    final Node node = addWaiter(Node.SHARED);
    // 初始化失败标志
    boolean failed = true;
    try {
        // 初始化线程中断标志
        boolean interrupted = false;
        for (;;) {
            // 获取当前节点的前继节点
            final Node p = node.predecessor();
            // 若前继节点为头结点，则执行tryAcquireShared获取资源
            if (p == head) {
                int r = tryAcquireShared(arg);
                // 若获取资源成功，且有剩余资源，将自己设为头结点并唤醒后续的阻塞线程
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    // 如果中断标志位为真，则线程执行自我了断
                    if (interrupted)
                        selfInterrupt();
                    // 表征获取资源成功
                    failed = false;
                    return;
                }
            }
            // houldParkAfterFailedAcquire(p, node)根据前继节点判断是否阻塞当前节点的线程
            // parkAndCheckInterrupt()阻塞当前线程并检查线程是否被中断过，若被中断过，将interrupted置为true
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            // 放弃获取资源
            cancelAcquire(node);
    }
}

可以发现，doAcquireShared与独占模式下的acquireQueued大同小异，主要有2点不同：

  1. doAcquireShared将线程的自我中断操作放在了方法体内部；
  2. 当线程获取到资源后，doAcquireShared会将当前线程所在的节点设为头结点，若资源有剩余则唤醒后续节点，比acquireQueued多了个唤醒后续节点的操作。

上述方法体现了共享的本质，即当前线程吃饱了后，若资源有剩余，会招呼后面排队的来一起吃，好东西要大家一起分享嘛，哈哈。

下面具体看一下setHeadAndPropagate(Node, int)函数：

private void setHeadAndPropagate(Node node, int propagate) {
    // 记录原来的头结点，下面过程会用到
    Node h = head; 
    // 设置新的头结点
    setHead(node);
    
    // 如果资源还有剩余，则唤醒后继节点
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

可以看到，实际执行唤醒后继节点的方法是doReleaseShared()，继续追踪：

private void doReleaseShared() {
    // 自旋操作
    for (;;) {
        // 获取等待队列的头结点
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue; 
                // 唤醒后继节点的线程
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

释放资源(共享模式)

首先进入到方法入口：

public final boolean releaseShared(int arg) {
    // 尝试释放资源
    if (tryReleaseShared(arg)) {
        // 唤醒后继节点的线程
        doReleaseShared();
        return true;
    }
    return false;
}

同样的，tryReleaseShared(int)由继承AQS的自定义同步器来具体实现。

doReleaseShared()上节讲解setHeadAndPropagate已说明过，不再赘述。

至此，共享模式下的资源获取/释放就讲解完了，下面以一个具体场景来概括一下：

整个获取/释放资源的过程是通过传播完成的，如最开始有10个资源，线程A、B、C分别需要5、4、3个资源。

  • A线程获取到5个资源，其发现资源还剩余5个，则唤醒B线程；
  • B线程获取到4个资源，其发现资源还剩余1个，唤醒C线程；
  • C线程尝试取3个资源，但发现只有1个资源，继续阻塞；
  • A线程释放1个资源，其发现资源还剩余2个，故唤醒C线程；
  • C线程尝试取3个资源，但发现只有2个资源，继续阻塞；
  • B线程释放2个资源，其发现资源还剩余4个，唤醒C线程；
  • C线程获取3个资源，其发现资源还剩1个，继续唤醒后续等待的D线程；
  • ......

 

转载：https://www.jianshu.com/p/0f876ead2846

AQS

标签：data   自定义   name   不同   源代码   idt   try   href   else   

原文地址：https://www.cnblogs.com/zzm96/p/12945799.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！