精华内容
下载资源
问答
  • 多线程AQS

    2019-10-07 00:10:37
    我们在多线程编程中或多或少的会用到锁,最普遍的应用就是sychronized,但是其自身的非公平锁,可重入锁的特性,使其无法适应复杂的业务场景,例如公平锁,读写锁等等,而我们的jdk提供了AbstractQueuedSynchronizer...

    我们在多线程编程中或多或少的会用到锁,最普遍的应用就是sychronized,但是其自身的非公平锁,可重入锁的特性,使其无法适应复杂的业务场景,例如公平锁,读写锁等等,而我们的jdk提供了AbstractQueuedSynchronizer(此后简称AQS),使我们可以进行扩展实现不同类型的锁以满足业务场景的需求。JDK提供的CountDownLatch,CyclicBarrier,ReentrantLock,Readwritelock,Semaphore,以及Condition都用到了类似AQS这个抽象类,这里我们就为大家解读下AQS源码,并从源码角度分析以上这几种锁的实现方式。废话不多说直接上图:
    在这里插入图片描述
    其实变量并不多可以大致分为3类,head,tail,链表的元素,每个线程都可以是该链表的一个节点,state锁的状态,unsafe及其以下的成员变量是通过反射直接内存数据的操作。而其中大量的方法无非也就是对链表的操作和state的修改,而state可以是重入锁的重入次数,也可以使读写锁读锁数量和写锁数量(前八位读数后八位写数,总之state是万能的),而为了满足锁的原子性采用的一些CAS操作。
    Condition 替代了传统Lock,Lock的wait notify需要配合sychronized使用,而Condition省去了这些还进行了一些扩展入notifyAll等。
    ReentrantLock可重入锁,又称作独占锁,一个标识位标识线程,一个标识位标识重入次数,只有线程标识位为空其他线程才可以获取锁。
    Readwritelock读写锁,分为两部分读锁和写锁。读锁为共享锁,写锁为独占锁。获取写锁时先判断读锁数量如果>0无法获取,获取读锁时如果写锁数量>0并且读锁不是当前线程无法获取,然后进入等待队列直到被唤醒,在编程时候注意锁降级(在释放写锁前先获取读锁可以节省不必要的开销)。
    在CountDownLatch中,此时state用来标记通信线程的数量,只有指定数量的的线程都countdown()后才会释放锁,此时循环锁也就是对外暴露的wait()所在的线程才会继续执行。
    CyclicBarrier采用ReentrantLock和condition实现分批次运行,只有指定数量的线程都执行到await()方法后才会继续执行,其余线程进入等待队列。
    Semaphore又叫令牌桶,时常高并发访问中被用作限流同一时间只能有指定数量的线程执行,其余线程呗放入等待队列。

    展开全文
  • 在Java多线程:线程间通信之Lock中我们提到了ReentrantLock是API级别的实现,但是...AQS本身仅仅是一个框架,定义了一套多线程访问共享资源的同步框架,可以实现ReentrantLock, Semaphore, CountDownLatch等多线程类...

    在Java多线程:线程间通信之Lock中我们提到了ReentrantLock是API级别的实现,但是没有说明其具体实现原理。实际上,ReentrantLock的底层实现使用了AQS(AbstractQueueSynchronizer)。AQS本身仅仅是一个框架,定义了一套多线程访问共享资源的同步框架,可以实现ReentrantLock, Semaphore, CountDownLatch等多线程类。

    AQS框架维护了一个资源state(volatile int)和一个同步队列。其中对state的访问包括三种方法:getState(), setState(), compareAndSetState()。其中,compareAndSetState()是原子操作,底层是CAS实现。

    AQS框架包含两种可供选择的实现方式:独占(Exclusive)和共享(Share)。由于不同自定义同步器征用共享资源的方式不同,自定义同步器实现时只需实现共享资源state的获取与释放方式即可,而不需要考虑队列的维护。下面简述AQS框架中独占锁和共享锁的获取,释放流程。

    独占锁流程

    获取时首先调用acquire(acquires),之后进入tryAcquire(acquires)尝试获取锁,若成功则返回。若失败则将当前线程构造为Node节点,CAS插入到同步队列尾部,该线程自旋。自旋时判断其前驱节点是否为头节点,是否成功获取同步状态,二者皆成立则当前节点设置为头节点,否则挂起当前线程等待被前驱节点唤醒。

    释放时首先调用release(acquires),之后进入tryRelease(acquires)释放同步状态,之后获取同步队列中当前节点的下一节点并唤醒。

    共享锁流程

    获取时首先调用acquireShared(acquires),之后进入tryAcquireShared(acquires)获取同步状态,返回值不小于0则说明同步状态有剩余,获取成功直接返回。若返回值小于0则说明获取同步状态失败,构造Node节点CAS插入同步队列尾部并自旋检查前驱节点是否为头节点且成功获取同步状态,若是则当前节点设为头节点,否则挂起等待被前驱节点唤醒。

    释放时调用releaseShared(acquires)释放同步状态,之后遍历整个队列唤醒所有后继节点。

    独占锁和共享锁实现区别

    独占锁的state值为1,同一时刻只有一个线程成功获取同步状态。共享锁state>1,取值由自定义同步器决定。

    独占锁队列头节点运行完毕释放锁后唤醒直接后继节点,共享锁唤醒所有后继节点。

    共享锁会出现多个线程同时成功获取同步状态的情况。

    重入锁的实现

    Java中的ReentrantLock和synchronized都是可重入锁,synchronized由JVM实现,重入锁实现时最主要的逻辑是判断上次获取锁的线程是否为当前线程,ReentrantLock基于AQS实现,提供公平锁和非公平锁两种方式,非公平锁实现逻辑如下:

    final boolean nonfairTryAcquire(int acquires) {

    //获取当前线程

    final Thread current = Thread.currentThread();

    //通过AQS获取同步状态

    int c = getState();

    //同步状态为0,说明临界区处于无锁状态,

    if (c == 0) {

    //修改同步状态,即加锁

    if (compareAndSetState(0, acquires)) {

    //将当前线程设置为锁的owner

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

    }

    读写锁的实现

    Java的ReentrantReadWriteLock是读写锁实现,其原理是将state变量的高16位和低16位拆分,高16位表示读锁,低16位表示写锁。其写锁tryAcquire(acquires)实现如下:

    获取同步状态,分离出低16位的写锁状态。

    同步状态不为0,则存在读锁或写锁。

    若存在读锁,则不能获取写锁。

    若当前线程不是上次获取写锁的线程,则不能获取写锁。

    以上判断通过,对低16位(写锁同步状态)进行CAS修改。

    当前线程设为写锁的获取线程。

    其读锁的tryAcquire(acquires)实现如下:

    获取当前同步状态,计算高16位为读锁状态+1后的值。

    若大于能获取到的读锁的最大值,则抛出异常。

    若存在写锁且当前线程不是写锁获取者,则获取读锁失败。

    若上述判断都通过,则利用CAS重新设置读锁的同步状态。

    写写锁释放与普通独占锁基本相同,在写锁释放中不断减少读锁的同步状态,同步状态为0时才能完全释放;读锁释放过程中不断释放写锁状态,直到为0,表示没有线程获取读锁。

    参考文献

    展开全文
  • Java多线程AQS

    2020-06-25 11:06:49
    队列同步器AbstractQueuedSynchronizer(以下简称同步器或AQS),是用来构建锁或者其他同步组件的基础框架,它使用了一个int成员变量表示同步状态,通过内置的FIFO队列来完成资源获取线程的排队工作。并发包的大师...

    AbstractQueuedSynchronizer

    学习AQS的必要性

    队列同步器AbstractQueuedSynchronizer(以下简称同步器或AQS),是用来构建锁或者其他同步组件的基础框架,它使用了一个int成员变量表示同步状态,通过内置的FIFO队列来完成资源获取线程的排队工作。并发包的大师(Doug Lea)期望它能够成为实现大部分同步需求的基础。

    AQS使用方式和其中的设计模式

    AQS的主要使用方式是继承,子类通过继承AQS并实现它的抽象方法来管理同步状态,在AQS里由一个int型的state来代表这个状态,在抽象方法的实现过程中免不了要对同步状态进行更改,这时就需要使用同步器提供的3个方法(getState()、setState(int newState)和compareAndSetState(int expect,int update))来进行操作,因为它们能够保证状态的改变是安全的。
    在这里插入图片描述
    在实现上,子类推荐被定义为自定义同步组件的静态内部类,AQS自身没有实现任何同步接口,它仅仅是定义了若干同步状态获取和释放的方法来供自定义同步组件使用,同步器既可以支持独占式地获取同步状态,也可以支持共享式地获取同步状态,这样就可以方便实现不同类型的同步组件(ReentrantLock、ReentrantReadWriteLock和CountDownLatch等)。

    同步器是实现锁(也可以是任意同步组件)的关键,在锁的实现中聚合同步器。可以这样理解二者之间的关系:
    锁是面向使用者的,它定义了使用者与锁交互的接口(比如可以允许两个线程并行访问),隐藏了实现细节;
    同步器面向的是锁的实现者,它简化了锁的实现方式,屏蔽了同步状态管理、线程的排队、等待与唤醒等底层操作。锁和同步器很好地隔离了使用者和实现者所需关注的领域。

    实现者需要继承同步器并重写指定的方法,随后将同步器组合在自定义同步组件的实现中,并调用同步器提供的模板方法,而这些模板方法将会调用使用者重写的方法。

    模板方法模式

    同步器的设计基于模板方法模式。模板方法模式的意图是,定义一个操作中的算法的骨架,而将一些步骤的实现延迟到子类中。模板方法使得子类可以不改变一个算法的结构即可重定义该算法的某些特定步骤。我们最常见的就是Spring框架里的各种Template。

    实际例子

    我们开了个蛋糕店,蛋糕店不能只卖一种蛋糕呀,于是我们决定先卖奶油蛋糕,芝士蛋糕和慕斯蛋糕。三种蛋糕在制作方式上一样,都包括造型,烘焙和涂抹蛋糕上的东西。所以可以定义一个抽象蛋糕模型
    在这里插入图片描述
    然后就可以批量生产三种蛋糕
    在这里插入图片描述
    在这里插入图片描述
    在这里插入图片描述
    在这里插入图片描述
    这样一来,不但可以批量生产三种蛋糕,而且如果日后有扩展,只需要继承抽象蛋糕方法就可以了,十分方便,我们天天生意做得越来越赚钱。突然有一天,我们发现市面有一种最简单的小蛋糕销量很好,这种蛋糕就是简单烘烤成型就可以卖,并不需要涂抹什么食材,由于制作简单销售量大,这个品种也很赚钱,于是我们也想要生产这种蛋糕。但是我们发现了一个问题,抽象蛋糕是定义了抽象的涂抹方法的,也就是说扩展的这种蛋糕是必须要实现涂抹方法,这就很鸡儿蛋疼了。怎么办?我们可以将原来的模板修改为带钩子的模板。
    在这里插入图片描述
    做小蛋糕的时候通过flag来控制是否涂抹,其余已有的蛋糕制作不需要任何修改可以照常进行。
    在这里插入图片描述

    AQS中的方法

    模板方法

    实现自定义同步组件时,将会调用同步器提供的模板方法,
    在这里插入图片描述
    这些模板方法同步器提供的模板方法基本上分为3类:独占式获取与释放同步状态、共享式获取与释放、同步状态和查询同步队列中的等待线程情况。

    可重写的方法

    在这里插入图片描述

    访问或修改同步状态的方法

    重写同步器指定的方法时,需要使用同步器提供的如下3个方法来访问或修改同步状态。
    •getState():获取当前同步状态。
    •setState(int newState):设置当前同步状态。
    •compareAndSetState(int expect,int update):使用CAS设置当前状态,该方法能够保证状态设置的原子性。

    CLH队列锁

    CLH队列锁即Craig, Landin, and Hagersten (CLH) locks。

    CLH队列锁也是一种基于链表的可扩展、高性能、公平的自旋锁,申请线程仅仅在本地变量上自旋,它不断轮询前驱的状态,假设发现前驱释放了锁就结束自旋。

    当一个线程需要获取锁时:
    1.创建一个的QNode,将其中的locked设置为true表示需要获取锁,myPred表示对其前驱结点的引用
    在这里插入图片描述

    2.线程A对tail域调用getAndSet方法,使自己成为队列的尾部,同时获取一个指向其前驱结点的引用myPred
    在这里插入图片描述
    线程B需要获得锁,同样的流程再来一遍
    在这里插入图片描述
    3.线程就在前驱结点的locked字段上旋转,直到前驱结点释放锁(前驱节点的锁值 locked == false)
    4.当一个线程需要释放锁时,将当前结点的locked域设置为false,同时回收前驱结点
    在这里插入图片描述
    如上图所示,前驱结点释放锁,线程A的myPred所指向的前驱结点的locked字段变为false,线程A就可以获取到锁。

    CLH队列锁的优点是空间复杂度低(如果有n个线程,L个锁,每个线程每次只获取一个锁,那么需要的存储空间是O(L+n),n个线程有n个myNode,L个锁有L个tail)。CLH队列锁常用在SMP体系结构下。

    Java中的AQS是CLH队列锁的一种变体实现。

    展开全文
  •  类如其名,抽象的队列式的同步器,AQS定义了一套多线程访问共享资源的同步器框架,许多同步类实现都依赖于它,如常用的ReentrantLock/Semaphore/CountDownLatch...。  以下是本文的目录大...

    一、概述

      谈到并发,不得不谈ReentrantLock;而谈到ReentrantLock,不得不谈AbstractQueuedSynchronizer(AQS)!

      类如其名,抽象的队列式的同步器,AQS定义了一套多线程访问共享资源的同步器框架,许多同步类实现都依赖于它,如常用的ReentrantLock/Semaphore/CountDownLatch...。

      以下是本文的目录大纲:

      1. 概述
      2. 框架
      3. 源码详解
      4. 简单应用

      若有不正之处,请谅解和批评指正,不胜感激。 

    二、框架

      它维护了一个volatile int state(代表共享资源)和一个FIFO线程等待队列(多线程争用资源被阻塞时会进入此队列)。这里volatile是核心关键词,具体volatile的语义,在此不述。state的访问方式有三种:

    • getState()
    • setState()
    • compareAndSetState()

      AQS定义两种资源共享方式:Exclusive(独占,只有一个线程能执行,如ReentrantLock)和Share(共享,多个线程可同时执行,如Semaphore/CountDownLatch)。

      不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法:

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

      以ReentrantLock为例,state初始化为0,表示未锁定状态。A线程lock()时,会调用tryAcquire()独占该锁并将state+1。此后,其他线程再tryAcquire()时就会失败,直到A线程unlock()到state=0(即释放锁)为止,其它线程才有机会获取该锁。当然,释放锁之前,A线程自己是可以重复获取此锁的(state会累加),这就是可重入的概念。但要注意,获取多少次就要释放多么次,这样才能保证state是能回到零态的。

      再以CountDownLatch以例,任务分为N个子线程去执行,state也初始化为N(注意N要与线程个数一致)。这N个子线程是并行执行的,每个子线程执行完后countDown()一次,state会CAS减1。等到所有子线程都执行完后(即state=0),会unpark()主调用线程,然后主调用线程就会从await()函数返回,继续后余动作。

      一般来说,自定义同步器要么是独占方法,要么是共享方式,他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式,如ReentrantReadWriteLock。

    三、源码详解

      本节开始讲解AQS的源码实现。依照acquire-release、acquireShared-releaseShared的次序来。

    3.1 acquire(int)

      此方法是独占模式下线程获取共享资源的顶层入口。如果获取到资源,线程直接返回,否则进入等待队列,直到获取到资源为止,且整个过程忽略中断的影响。这也正是lock()的语义,当然不仅仅只限于lock()。获取到资源后,线程就可以去执行其临界区代码了。下面是acquire()的源码:

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

      函数流程如下:

      1. tryAcquire()尝试直接去获取资源,如果成功则直接返回;
      2. addWaiter()将该线程加入等待队列的尾部,并标记为独占模式;
      3. acquireQueued()使线程在等待队列中获取资源,一直获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
      4. 如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断selfInterrupt(),将中断补上。

      这时单凭这4个抽象的函数来看流程还有点朦胧,不要紧,看完接下来的分析后,你就会明白了。就像《大话西游》里唐僧说的:等你明白了舍生取义的道理,你自然会回来和我唱这首歌的。

    3.1.1 tryAcquire(int)

      此方法尝试去获取独占资源。如果获取成功,则直接返回true,否则直接返回false。这也正是tryLock()的语义,还是那句话,当然不仅仅只限于tryLock()。如下是tryAcquire()的源码:

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

      什么?直接throw异常?说好的功能呢?好吧,还记得概述里讲的AQS只是一个框架,具体资源的获取/释放方式交由自定义同步器去实现吗?就是这里了!!!AQS这里只定义了一个接口,具体资源的获取交由自定义同步器去实现了(通过state的get/set/CAS)!!!至于能不能重入,能不能加塞,那就看具体的自定义同步器怎么去设计了!!!当然,自定义同步器在进行资源访问时要考虑线程安全的影响。

      这里之所以没有定义成abstract,是因为独占模式下只用实现tryAcquire-tryRelease,而共享模式下只用实现tryAcquireShared-tryReleaseShared。如果都定义成abstract,那么每个模式也要去实现另一模式下的接口。说到底,Doug Lea还是站在咱们开发者的角度,尽量减少不必要的工作量。

    3.1.2 addWaiter(Node)

      此方法用于将当前线程加入到等待队列的队尾,并返回当前线程所在的结点。还是上源码吧:

    private Node addWaiter(Node mode) {
        //以给定模式构造结点。mode有两种:EXCLUSIVE(独占)和SHARED(共享)
        Node node = new Node(Thread.currentThread(), mode);
        
        //尝试快速方式直接放到队尾。
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        
        //上一步失败则通过enq入队。
        enq(node);
        return node;
    }

     不用再说了,直接看注释吧。这里我们说下Node。Node结点是对每一个访问同步代码的线程的封装,其包含了需要同步的线程本身以及线程的状态,如是否被阻塞,是否等待唤醒,是否已经被取消等。变量waitStatus则表示当前被封装成Node结点的等待状态,共有4种取值CANCELLED、SIGNAL、CONDITION、PROPAGATE。

    • CANCELLED:值为1,在同步队列中等待的线程等待超时或被中断,需要从同步队列中取消该Node的结点,其结点的waitStatus为CANCELLED,即结束状态,进入该状态后的结点将不会再变化。

    • SIGNAL:值为-1,被标识为该等待唤醒状态的后继结点,当其前继结点的线程释放了同步锁或被取消,将会通知该后继结点的线程执行。说白了,就是处于唤醒状态,只要前继结点释放锁,就会通知标识为SIGNAL状态的后继结点的线程执行。

    • CONDITION:值为-2,与Condition相关,该标识的结点处于等待队列中,结点的线程等待在Condition上,当其他线程调用了Condition的signal()方法后,CONDITION状态的结点将从等待队列转移到同步队列中,等待获取同步锁。

    • PROPAGATE:值为-3,与共享模式相关,在共享模式中,该状态标识结点的线程处于可运行状态。

    • 0状态:值为0,代表初始化状态。

    AQS在判断状态时,通过用waitStatus>0表示取消状态,而waitStatus<0表示有效状态。

    3.1.2.1 enq(Node)

       此方法用于将node加入队尾。源码如下:

    private Node enq(final Node node) {
        //CAS"自旋",直到成功加入队尾
        for (;;) {
            Node t = tail;
            if (t == null) { // 队列为空,创建一个空的标志结点作为head结点,并将tail也指向它。
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {//正常流程,放入队尾
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

     

    如果你看过AtomicInteger.getAndIncrement()函数源码,那么相信你一眼便看出这段代码的精华。CAS自旋volatile变量,是一种很经典的用法。还不太了解的,自己去百度一下吧。

    3.1.3 acquireQueued(Node, int)

      OK,通过tryAcquire()和addWaiter(),该线程获取资源失败,已经被放入等待队列尾部了。聪明的你立刻应该能想到该线程下一部该干什么了吧:进入等待状态休息,直到其他线程彻底释放资源后唤醒自己,自己再拿到资源,然后就可以去干自己想干的事了。没错,就是这样!是不是跟医院排队拿号有点相似~~acquireQueued()就是干这件事:在等待队列中排队拿号(中间没其它事干可以休息),直到拿到号后再返回。这个函数非常关键,还是上源码吧:

    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;//标记是否成功拿到资源
        try {
            boolean interrupted = false;//标记等待过程中是否被中断过
            
            //又是一个“自旋”!
            for (;;) {
                final Node p = node.predecessor();//拿到前驱
                //如果前驱是head,即该结点已成老二,那么便有资格去尝试获取资源(可能是老大释放完资源唤醒自己的,当然也可能被interrupt了)。
                if (p == head && tryAcquire(arg)) {
                    setHead(node);//拿到资源后,将head指向该结点。所以head所指的标杆结点,就是当前获取到资源的那个结点或null。
                    p.next = null; // setHead中node.prev已置为null,此处再将head.next置为null,就是为了方便GC回收以前的head结点。也就意味着之前拿完资源的结点出队了!
                    failed = false;
                    return interrupted;//返回等待过程中是否被中断过
                }
                
                //如果自己可以休息了,就进入waiting状态,直到被unpark()
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;//如果等待过程中被中断过,哪怕只有那么一次,就将interrupted标记为true
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

     

    到这里了,我们先不急着总结acquireQueued()的函数流程,先看看shouldParkAfterFailedAcquire()和parkAndCheckInterrupt()具体干些什么。

    3.1.3.1 shouldParkAfterFailedAcquire(Node, Node)

      此方法主要用于检查状态,看看自己是否真的可以去休息了(进入waiting状态,如果线程状态转换不熟,可以参考本人上一篇写的Thread详解),万一队列前边的线程都放弃了只是瞎站着,那也说不定,对吧!

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;//拿到前驱的状态
        if (ws == Node.SIGNAL)
            //如果已经告诉前驱拿完号后通知自己一下,那就可以安心休息了
            return true;
        if (ws > 0) {
            /*
             * 如果前驱放弃了,那就一直往前找,直到找到最近一个正常等待的状态,并排在它的后边。
             * 注意:那些放弃的结点,由于被自己“加塞”到它们前边,它们相当于形成一个无引用链,稍后就会被保安大叔赶走了(GC回收)!
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
             //如果前驱正常,那就把前驱的状态设置成SIGNAL,告诉它拿完号后通知自己一下。有可能失败,人家说不定刚刚释放完呢!
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

     

    整个流程中,如果前驱结点的状态不是SIGNAL,那么自己就不能安心去休息,需要去找个安心的休息点,同时可以再尝试下看有没有机会轮到自己拿号。

    3.1.3.2 parkAndCheckInterrupt()

      如果线程找好安全休息点后,那就可以安心去休息了。此方法就是让线程去休息,真正进入等待状态。

    1 private final boolean parkAndCheckInterrupt() {
    2     LockSupport.park(this);//调用park()使线程进入waiting状态
    3     return Thread.interrupted();//如果被唤醒,查看自己是不是被中断的。
    4 }

       park()会让当前线程进入waiting状态。在此状态下,有两种途径可以唤醒该线程:1)被unpark();2)被interrupt()。(再说一句,如果线程状态转换不熟,可以参考本人写的Thread详解)。需要注意的是,Thread.interrupted()会清除当前线程的中断标记位。 

    3.1.3.3 小结

      OK,看了shouldParkAfterFailedAcquire()和parkAndCheckInterrupt(),现在让我们再回到acquireQueued(),总结下该函数的具体流程:

    1. 结点进入队尾后,检查状态,找到安全休息点;
    2. 调用park()进入waiting状态,等待unpark()或interrupt()唤醒自己;
    3. 被唤醒后,看自己是不是有资格能拿到号。如果拿到,head指向当前结点,并返回从入队到拿到号的整个过程中是否被中断过;如果没拿到,继续流程1。

     

    3.1.4 小结

      OKOK,acquireQueued()分析完之后,我们接下来再回到acquire()!再贴上它的源码吧:

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

    再来总结下它的流程吧:

    1. 调用自定义同步器的tryAcquire()尝试直接去获取资源,如果成功则直接返回;
    2. 没成功,则addWaiter()将该线程加入等待队列的尾部,并标记为独占模式;
    3. acquireQueued()使线程在等待队列中休息,有机会时(轮到自己,会被unpark())会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
    4. 如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断selfInterrupt(),将中断补上。

    由于此函数是重中之重,我再用流程图总结一下:

    至此,acquire()的流程终于算是告一段落了。这也就是ReentrantLock.lock()的流程,不信你去看其lock()源码吧,整个函数就是一条acquire(1)!!!

     

    3.2 release(int)

       上一小节已经把acquire()说完了,这一小节就来讲讲它的反操作release()吧。此方法是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即state=0),它会唤醒等待队列里的其他线程来获取资源。这也正是unlock()的语义,当然不仅仅只限于unlock()。下面是release()的源码:

    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()来释放资源。有一点需要注意的是,它是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了!所以自定义同步器在设计tryRelease()的时候要明确这一点!!

    3.2.1 tryRelease(int)

      此方法尝试去释放指定量的资源。下面是tryRelease()的源码:

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

      跟tryAcquire()一样,这个方法是需要独占模式的自定义同步器去实现的。正常来说,tryRelease()都会成功的,因为这是独占模式,该线程来释放资源,那么它肯定已经拿到独占资源了,直接减掉相应量的资源即可(state-=arg),也不需要考虑线程安全的问题。但要注意它的返回值,上面已经提到了,release()是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了!所以自义定同步器在实现时,如果已经彻底释放资源(state=0),要返回true,否则返回false。

    3.2.2 unparkSuccessor(Node)

      此方法用于唤醒等待队列中下一个线程。下面是源码:

    private void unparkSuccessor(Node node) {
        //这里,node一般为当前线程所在的结点。
        int ws = node.waitStatus;
        if (ws < 0)//置零当前线程所在的结点状态,允许失败。
            compareAndSetWaitStatus(node, ws, 0);

        Node s = node.next;//找到下一个需要唤醒的结点s
        if (s == null || s.waitStatus > 0) {//如果为空或已取消
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)//从这里可以看出,<=0的结点,都是还有效的结点。
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);//唤醒
    }

     

      这个函数并不复杂。一句话概括:用unpark()唤醒等待队列中最前边的那个未放弃线程,这里我们也用s来表示吧。此时,再和acquireQueued()联系起来,s被唤醒后,进入if (p == head && tryAcquire(arg))的判断(即使p!=head也没关系,它会再进入shouldParkAfterFailedAcquire()寻找一个安全点。这里既然s已经是等待队列中最前边的那个未放弃线程了,那么通过shouldParkAfterFailedAcquire()的调整,s也必然会跑到head的next结点,下一次自旋p==head就成立啦),然后s把自己设置成head标杆结点,表示自己已经获取到资源了,acquire()也返回了!!And then, DO what you WANT!

    3.2.3 小结

      release()是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即state=0),它会唤醒等待队列里的其他线程来获取资源。

    3.3 acquireShared(int)

      此方法是共享模式下线程获取共享资源的顶层入口。它会获取指定量的资源,获取成功则直接返回,获取失败则进入等待队列,直到获取到资源为止,整个过程忽略中断。下面是acquireShared()的源码:

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

      这里tryAcquireShared()依然需要自定义同步器去实现。但是AQS已经把其返回值的语义定义好了:负值代表获取失败;0代表获取成功,但没有剩余资源;正数表示获取成功,还有剩余资源,其他线程还可以去获取。所以这里acquireShared()的流程就是:

      1. tryAcquireShared()尝试获取资源,成功则直接返回;
      2. 失败则通过doAcquireShared()进入等待队列,直到获取到资源为止才返回。

    3.3.1 doAcquireShared(int)

      此方法用于将当前线程加入等待队列尾部休息,直到其他线程释放资源唤醒自己,自己成功拿到相应量的资源后才返回。下面是doAcquireShared()的源码:

    private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);//加入队列尾部
        boolean failed = true;//是否成功标志
        try {
            boolean interrupted = false;//等待过程中是否被中断过的标志
            for (;;) {
                final Node p = node.predecessor();//前驱
                if (p == head) {//如果到head的下一个,因为head是拿到资源的线程,此时node被唤醒,很可能是head用完资源来唤醒自己的
                    int r = tryAcquireShared(arg);//尝试获取资源
                    if (r >= 0) {//成功
                        setHeadAndPropagate(node, r);//将head指向自己,还有剩余资源可以再唤醒之后的线程
                        p.next = null; // help GC
                        if (interrupted)//如果等待过程中被打断过,此时将中断补上。
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                
                //判断状态,寻找安全点,进入waiting状态,等着被unpark()或interrupt()
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

     

      有木有觉得跟acquireQueued()很相似?对,其实流程并没有太大区别。只不过这里将补中断的selfInterrupt()放到doAcquireShared()里了,而独占模式是放到acquireQueued()之外,其实都一样,不知道Doug Lea是怎么想的。

      跟独占模式比,还有一点需要注意的是,这里只有线程是head.next时(“老二”),才会去尝试获取资源,有剩余的话还会唤醒之后的队友。那么问题就来了,假如老大用完后释放了5个资源,而老二需要6个,老三需要1个,老四需要2个。老大先唤醒老二,老二一看资源不够,他是把资源让给老三呢,还是不让?答案是否定的!老二会继续park()等待其他线程释放资源,也更不会去唤醒老三和老四了。独占模式,同一时刻只有一个线程去执行,这样做未尝不可;但共享模式下,多个线程是可以同时执行的,现在因为老二的资源需求量大,而把后面量小的老三和老四也都卡住了。当然,这并不是问题,只是AQS保证严格按照入队顺序唤醒罢了(保证公平,但降低了并发)。

     

    3.3.1.1 setHeadAndPropagate(Node, int)

    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; 
        setHead(node);//head指向自己
         //如果还有剩余量,继续唤醒下一个邻居线程
        if (propagate > 0 || h == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

     

      此方法在setHead()的基础上多了一步,就是自己苏醒的同时,如果条件符合(比如还有剩余资源),还会去唤醒后继结点,毕竟是共享模式!

      doReleaseShared()我们留着下一小节的releaseShared()里来讲。

     

    3.3.2 小结

      OK,至此,acquireShared()也要告一段落了。让我们再梳理一下它的流程:

      1. tryAcquireShared()尝试获取资源,成功则直接返回;
      2. 失败则通过doAcquireShared()进入等待队列park(),直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

      其实跟acquire()的流程大同小异,只不过多了个自己拿到资源后,还会去唤醒后继队友的操作(这才是共享嘛)

    3.4 releaseShared()

      上一小节已经把acquireShared()说完了,这一小节就来讲讲它的反操作releaseShared()吧。此方法是共享模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果成功释放且允许唤醒等待线程,它会唤醒等待队列里的其他线程来获取资源。下面是releaseShared()的源码:

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

     

      此方法的流程也比较简单,一句话:释放掉资源后,唤醒后继。跟独占模式下的release()相似,但有一点稍微需要注意:独占模式下的tryRelease()在完全释放掉资源(state=0)后,才会返回true去唤醒其他线程,这主要是基于独占下可重入的考量;而共享模式下的releaseShared()则没有这种要求,共享模式实质就是控制一定量的线程并发执行,那么拥有资源的线程在释放掉部分资源时就可以唤醒后继等待结点。例如,资源总量是13,A(5)和B(7)分别获取到资源并发运行,C(4)来时只剩1个资源就需要等待。A在运行过程中释放掉2个资源量,然后tryReleaseShared(2)返回true唤醒C,C一看只有3个仍不够继续等待;随后B又释放2个,tryReleaseShared(2)返回true唤醒C,C一看有5个够自己用了,然后C就可以跟A和B一起运行。而ReentrantReadWriteLock读锁的tryReleaseShared()只有在完全释放掉资源(state=0)才返回true,所以自定义同步器可以根据需要决定tryReleaseShared()的返回值。

    3.4.1 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;
            }
            if (h == head)// head发生变化
                break;
        }
    }

     

    3.5 小结

      本节我们详解了独占和共享两种模式下获取-释放资源(acquire-release、acquireShared-releaseShared)的源码,相信大家都有一定认识了。值得注意的是,acquire()和acquireShared()两种方法下,线程在等待队列中都是忽略中断的。AQS也支持响应中断的,acquireInterruptibly()/acquireSharedInterruptibly()即是,这里相应的源码跟acquire()和acquireShared()差不多,这里就不再详解了。

     

    四、简单应用

      通过前边几个章节的学习,相信大家已经基本理解AQS的原理了。这里再将“框架”一节中的一段话复制过来:

      不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法:

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

      OK,下面我们就以AQS源码里的Mutex为例,讲一下AQS的简单应用。

    4.1 Mutex(互斥锁)

      Mutex是一个不可重入的互斥锁实现。锁资源(AQS里的state)只有两种状态:0表示未锁定,1表示锁定。下边是Mutex的核心源码:

    class Mutex implements Lock, java.io.Serializable {
        // 自定义同步器
        private static class Sync extends AbstractQueuedSynchronizer {
            // 判断是否锁定状态
            protected boolean isHeldExclusively() {
                return getState() == 1;
            }

            // 尝试获取资源,立即返回。成功则返回true,否则false。
            public boolean tryAcquire(int acquires) {
                assert acquires == 1; // 这里限定只能为1个量
                if (compareAndSetState(0, 1)) {//state为0才设置为1,不可重入!
                    setExclusiveOwnerThread(Thread.currentThread());//设置为当前线程独占资源
                    return true;
                }
                return false;
            }

            // 尝试释放资源,立即返回。成功则为true,否则false。
            protected boolean tryRelease(int releases) {
                assert releases == 1; // 限定为1个量
                if (getState() == 0)//既然来释放,那肯定就是已占有状态了。只是为了保险,多层判断!
                    throw new IllegalMonitorStateException();
                setExclusiveOwnerThread(null);
                setState(0);//释放资源,放弃占有状态
                return true;
            }
        }

        // 真正同步类的实现都依赖继承于AQS的自定义同步器!
        private final Sync sync = new Sync();

        //lock<-->acquire。两者语义一样:获取资源,即便等待,直到成功才返回。
        public void lock() {
            sync.acquire(1);
        }

        //tryLock<-->tryAcquire。两者语义一样:尝试获取资源,要求立即返回。成功则为true,失败则为false。
        public boolean tryLock() {
            return sync.tryAcquire(1);
        }

        //unlock<-->release。两者语文一样:释放资源。
        public void unlock() {
            sync.release(1);
        }

        //锁是否占有状态
        public boolean isLocked() {
            return sync.isHeldExclusively();
        }
    }

     

      同步类在实现时一般都将自定义同步器(sync)定义为内部类,供自己使用;而同步类自己(Mutex)则实现某个接口,对外服务。当然,接口的实现要直接依赖sync,它们在语义上也存在某种对应关系!!而sync只用实现资源state的获取-释放方式tryAcquire-tryRelelase,至于线程的排队、等待、唤醒等,上层的AQS都已经实现好了,我们不用关心。

      除了Mutex,ReentrantLock/CountDownLatch/Semphore这些同步类的实现方式都差不多,不同的地方就在获取-释放资源的方式tryAcquire-tryRelelase。掌握了这点,AQS的核心便被攻破了!

      OK,至此,整个AQS的讲解也要落下帷幕了。希望本文能够对学习Java并发编程的同学有所借鉴,中间写的有不对的地方,也欢迎讨论和指正~

    展开全文
  • 03.并发多线程AQS

    2019-08-01 17:29:36
    AQS是什么 全称是AbstractQueuedSynchronizer,(抽象队列同步器) 它的定位手机为java中几乎所有的锁和同步器提供一个基础框架 AQS是基于FIFO的队列(First-In,First-Out,先进先出)实现的,并且内部维护了一个...
  • } } 输出: Thread-0 释放1把锁 Thread-1 释放1把锁 一直等待 分配的3把锁全部释放,main 线程继续运行 /** * 分配的3把锁全部释放,main 线程继续运行 */ public static void allReleaseTest() { //分配3把共享锁 ...
  • ![图片说明](https://img-ask.csdn.net/upload/201908/05/1564936762_969108.jpg) ![图片说明](https://img-ask.csdn.net/upload/201908/05/1564936770_291455.jpg) ![图片说明]...
  • 我们前面几张提到过,JUC 这个包里面的工具类的底层就是使用 CAS 和 volatile 来保证线程安全的,整个 JUC 包里面的类都是基于它们构建的。今天我们介绍一个非常重要的同步器,这个类是 JDK 在 CAS 和 volatile 的...
  • 同步器内部依赖一个FIFO的双向队列来完成资源获取线程的排队工作。 2.同步器的应用  同步器主要使用方式是继承,子类通过继承同步器并实现它的抽象方法来管理同步状态,对同步状态的修改或者访问主要通过同步器...
  • 共享只读 : 一个共享只读的对象,在没有额外同步的情况下,可以被线程并发访问,但是任何线程都不能修改它3.线程安全对象 : 一个线程安全的对象或则容器,在内部通过同步机制来保证线程安全,所以其他线程...
  • 首先,为什么要理解AQS??? 因为同步组件(这里不仅仅指锁,还包括CountDownLatch等)的实现依赖于同步器AQS,即AQS是同步组件实现的核心部分。 那么,AQS到底是什么呢??? AQS(AbstractQueuedSynchronizer),...
  • } tryAcquireShared(arg) 方法由AQS的子类Sync 的子类NonfairSync 实现 public class Semaphore implements java.io.Serializable { ... static final class NonfairSync extends Sync { private static final long...
  • Java.util.concurrent 是在并发编程中比较常用的工具类,里面包含很用来在并发场景中使用的组件。比如线程池、阻塞队列、计时器、同步器、并发集合等等 Lock Lock 在 J.U.C 中是最核心的组件 lock的实现 Lock 本质...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 2,887
精华内容 1,154
关键字:

多线程aqs