美文网首页
从ReentrantLock的实现看AQS的原理及应用

从ReentrantLock的实现看AQS的原理及应用

作者: 星空怎样 | 来源:发表于2020-05-28 17:54 被阅读0次

    [toc]

    前言

    Java的大部分的同步类(Lock、Semaphore、ReentrantLock等)都是基于AbstractQueueSynchronizer(简称AQS)实现的。AQS是一种提供了原子式管理同步状态、阻塞和唤醒线程功能以及队列模型的简单框架,本文会从应用层逐渐深入到原理层,并通过ReentrantLock的基本特性和ReentrantLock与AQS关联,来深入解读AQS相关独占锁的知识点,同时采取问答模式来帮助理解AQS,本文主要阐述AQS中独占锁的缩进和Sync Queue,不讲述包含共享锁和Condition Queue部分(本文核心为AQS原理剖析,简单介绍了ReentraantLock),下面是本篇文章的大纲和思路:

    image

    1.ReentrantLock

    1.1ReentrrantLock特性概述

    ReentrantLock意思是可重入锁,指的是一个线程能够对一个临界资源重复加锁,为了帮助大家更好理解ReentrantLock的特性,我们先将ReentranLock和synchronized进行比较,其特性如下(蓝色部分是本篇文章主要剖析的的点):

    image

    下面通过伪代码,进行更加直观的比较

    //使用synchronized
    //1、用于代码块
    synchronized(this){}
    //用于对象
    synchronized(object){}
    //用于方法
    public synchronized void test(){}
    //可重入
    for(;;){
        synchronized(this){}
    }
    
    //使用ReentrantLock的使用方式
    public void test(){
        //选择是否公平,true:公平,false:不公平,默认是false
        ReentrantLock lock=new ReentrantLock(true);
        lock.lock();
        try{
            try {
                // 3.支持多种加锁方式,比较灵活; 具有可重入特性
                if(lock.tryLock(100, TimeUnit.MILLISECONDS)){ }
            } finally {
                // 4.手动释放锁
                lock.unlock()
            }
        }finally{
            lock.unlock();
        }
    }
    

    1.2ReentrantLock与AQS的关联

    通过上文了解,ReentrantLock支持公平锁和非公平锁,并且ReentrantLock的底层就是由AQS来实现的。那么ReentrantLock是如何通过公平锁和非公平锁与AQS关联起来呢?我们着重从这两者的几所过程来理解一下他们与AQS之间的关系(加锁过程中与AQS关联比较明显,解锁流程后续介绍)。

    下面是非公平锁的加锁源码:

    static final class NonfairSync extends Sync {
            ...
            final void lock() {
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    acquire(1);
            }
    
            ...
        }
    

    代码含义:

    • 通过CAS设置变量state(同步状态)成功,也就是获取锁成功,则将当前线程设置为独占线程
    • 通过CAS设置变量state(同步状态)失败,也就是获取锁失败,则进入Acquire方法进行后续处理

    第一步很好理解,但第二步获取失败后,后续处理策略时怎样的呢?这块可能会有以下思考:

    • 某个线程获取锁失败的后续流程是什么,以下两种可能。
      • 1:将当前线程获取结果设置为失败,获取锁流程结束,这种设计极大降低系统的并发度,并不满足实际需求,所以需要下面这种流程,也就是AQS框架的处理流程。
      • 2、存在某种排队等候机制,线程继续等待然然保留获取锁的可能,获取锁的流程仍在继续。
        • 既然说到了排队等候机制,那么就一定会具有某种队列形式,这样的队列是什么数据结构呢?
        • 处于排队等候机制的线程,什么时候可以有机会获取锁?
        • 如果处于排队等候机制中的线程一直无法获取锁,还是需要一直等待吗?还有别的策略吗?

    带着非公平锁的这些问题。在看一下公平锁的源码:

     static final class FairSync extends Sync {
           ...
            final void lock() {
                acquire(1);
            }
         ...
     }
    

    看这块代码,我们可能会存在这种疑问,Lock函数通过Acquire方法进行加锁,但是具体是如何加锁的呢?

    结合公平锁和非公平锁的加锁流程,虽然流程上有一定的不同,但是都调用了acquire()方法,而acquire()方法是FairSync和NonfairSync的父类AQS的核心方法。

    对于上面提到的问题,其实在ReentrantLock源码中无法解答,而这些问题位于acquire方法所在类AQS中,也就是本文核心---AQS。下面我们会对AQS以及ReentrantLock和AQS的关联做详细介绍(相关问题在2.35小节下面讲解)

    2.AQS

    首先通过下面的框架图整体了解一下AQS框架:(图中API层第三列最后一个tryAcquireSharedNanos是共享模式忽略中断,图中标注写错了)


    image
    • 上图中有颜色的为Method,无严肃的是Attribution
    • 总的来说,AQS框架分为五层,自上而下由浅入深,从AQS对象暴露的API到底层基础数据。
    • 当有自定义同步器接入时,只需要重写第一层所需要的部分方法即可,不需要关注底层的实现流程,当自定义同步进行加锁或者解锁操作时,先经过第一层的API进入到AQS内部方法,然后经过第二次进行锁的获取,接着对于获取锁失败的流程,进入第三层和第四层,而这些方式均依赖于第五层的基础数据提供层。

    下面我们会从整体到细节,从流程到方法逐一剖析AQS框架,主要分析过程如下:


    image

    2.1 原理概览

    AQS的核心思想是,如果被请求的共享资源空闲,那么就将当前请求资源的线程设置为有效的工作线程,将共享资源设置为锁定状态;如果共享资源被占用,就需要一定的阻塞等待来保证锁的分配,这个机制主要用的是CLH队列的变体实现的,将暂时获取不到的锁的队列加入到队列中。

    CLH;Craig、Landin and Hagerrsten队列是单向链表,AQS中的队列是CLH变体的虚拟双向队列(FIFO),AQS是通过将没条请求共享资源按封装成一个节点来实现锁的分配。 如下图:

    image

    AQS使用一个volatile的int类型的成员变量来表示同步状态,通过内置的FIFO队列来完成资源获取的排队工作,通过CAS完成对state值的修改

    2.1.1 AQS数据结构

    先来看下AQS中最基本的数据结构Node,Node即为上面的CLH变体队列中的节点:

    image

    解释一下几个方法属性值的含义:

    方法和属性 含义
    waitStatus 当前节点在队列中的状态
    thread 表示处于该节点的线程
    prv 前驱指针
    predecessor 返回前驱节点,如果没有抛出空指针异常
    nexWaiter 指向下一个处于CONDITION状态的节点
    next 后继指针

    线程两种锁模式:

    模式 含义
    SHARED 表示线程以共享的模式等待锁
    EXCLUSIVE 表示线程正在以独占方式等待锁

    waitStatus有下面几个枚举值:

    枚举 含义
    0 当一个Node被初始化时候的默认值
    CANCELLED 为1,表示线程获取锁的请求已经取消了
    CONDITION 为-2,表示节点在等待队列中,节点线程等待唤醒
    PROPAGATE 为-3,当前线程处于一个SHARED情况下,该字段才会使用
    SIGNAL 为-1,表示线程已经准备好了,就等资源释放了

    2.1.2 同步状态state

    在了解数据结构后,接下来了解一下AQS的同步状态state。AQS中维护了一个名为state的字段,意思是同步状态,是由volatile修饰的,用于展示当前临界资源的获取情况。

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    //在读写锁中,高16位表示共享锁的标志,低16位表示独占锁
    private volatile int state;
    

    下面提供您了几个访问这个字段的方法:

    方法名 描述
    protected final int getState() 获取state的值
    protected final void setState(int newState) 设置state的值
    protected final boolean compareAndSetState(int expect,int update) 使用CAS方式更新state

    这几个方法都是final修饰的,说明子类无法重写他们,我们可以通过state字段表示的同步状态来实现多线程的独占模式和共享模式(加锁过程)。

    image image
    对于我们自定义的同步工具。需要自定义获取同步状态和释放状态的方式,也就是AQS第一层:API层

    2.2 AQS的重要方法与ReentrantLock的关联

    从架构图中可以得知,AQS提供了大量的用于自定义同步器实现的protested方法,自定义同步器实现的相关方法只是为了通过修改state字段来实现多线程的独占模式和共享模式。自定义同步器需要实现以下方法(ReentrantLock需要实现的方法如下,并不是全部):

    方法名 描述
    protected boolean isHeldExclusively 该线程是否正在独占资源。只有用到Conditionx才需要实现他
    protected boolean tryAcquire(int arg) 独占方式。arg表示获取锁的次数,尝试获取资源,成功则返回true,失败返回false
    protected boolean tryRelease(int arg) 独占方式。arg为释放锁的次数,尝试释放资源,成功,则返回true,失败则返回false
    protected int tryAcquireShared(int arg) 共享方式,arg为获取锁的次数,尝试获取资源,负数表示失败,0表示成功,但没有剩余可用资源,正式表示成功,且有剩余资源
    protected boolean tryReleaseShard(int arg) 共享方式。arg为释放锁的次数,尝试释放资源,如果释放后运行唤醒后续等待节点返回true,否则返回false

    一般来说,自定义同步器要么是独占方式,要么是共享方式,他们只需要tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可(ReentrantLock因为是独占的所以实现了tryAcquire-tryRelease),AQS也指出自定义同步器可以同时实现独占锁和排它锁,如ReentrantReadWriteLock,而ReentrantLock是独占锁所以只实现了tryAcquire-tryRelease。

    非公平锁为例,这里阐述一下非公平锁与AQS方法的关联之处,具体每一处核心方法的作用会在文章后面详细进行阐述:

    image

    为了帮助大家理解ReentrantLock和AQS之间方法的交互过程,以非公平锁为例,我们将加锁和解锁的交互流程单独拎出来强调一下,以便于对后续内容的理解:

    image

    加锁过程:

    • 加锁通过ReentrantLock的加锁方法lock进行加锁操作。
    • 会调用到内部类Sync的lock方法,由于Sync的lock方法是抽象方法,根据ReenTrantLock初始化选择的是非公平锁和公平锁,执行相关内部类的lock,本质上都会执行AQS的acquire方法。
    • AQS的acquire方法会执行tryAcquire方法,但是由于tryAcquire需要自定义同步器实现,因此执行了ReentrantLock中的tryAccquire方法,由于ReentrantLock是通过公平锁和非公平锁内部类实现的tryAcquire方法,因此根据锁类型不同,执行不同的tryAcquire。
    • tryAcquire是获取锁逻辑,获取失败后,会执行AQS的后续逻辑,根ReentrantLock自定义同步器无关。

    加锁过程:

    • 通过ReentrantLock的解锁方法unLock进行解锁
    • unlock会调用内部类Sync的release方法,该方法继承于AQS
    • release会调用tryRelease方法,tryRelease需要自定义同步器实现,tryRelease只在ReentrantLock中的Sync实现,因此可以看出,释放锁的过程,并区分是否公平锁。
    • 释放成功后,所有处理由AQS框架完成,与自定义同步器无关。

    通过上面的描述,大概可以总结出ReentrantLock加锁解锁的时API层核心方法的映射关系。

    image

    2.3 通过ReentranLock理解的AQS

    ReentrantLock中公平锁和非公平锁的底层实现是相同的,这里以非公平锁为例进行分析。

    在非公平锁中,有一段代码:

    // java.util.concurrent.locks.ReentrantLock
    
    static final class NonfairSync extends Sync {
        ...
        final void lock() {
            //因为是非公平锁,所以先尝试去获取锁,如果成功了那么直接将当前线程持有锁,如果失败了调用AQS的acquire方法,所以非公平锁会抢占两次,第一次是在第一次锁时候进行抢占,第二次是加入到对象中的时候也会抢占一次
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
        ...
    }
    

    看这个acquire是怎么写的:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    public final void acquire(int arg) {
        //加入队列,如果是非公平锁会再一次进行抢占,如果是公平锁,会放到队列尾部的
        if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    

    在看tryAcquire方法:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    //该方法需要子类去实现
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }
    

    可以看出,这里是AQS的简单实现,具体获取锁的实现方法是由各自的公平锁和非公平锁单独实现的(以ReentrantLock为例),如果该方法返回了true,则说明当前线程获取锁成功,就不用往后执行了,如果获取失败,就需要加入等待队列中,下面详细介绍何时以及怎样被加入等待队列中的。

    2.3.1 线程加入等待队列

    2.3.1.1 加入队列的时机

    当执行acquire(1)时会通过,tryAcquire获取锁,在这种情况下,如果获取锁失败,就会调用addWaiter加入到等待队列中去。

    2.3.1.2 如何加入队列

    获取锁失败后,会执行addWaiter(Node.EXCLUSIVE)加入等待队列中,具体实现方法如下:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    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;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        //如果pred是空说明之前尾结点是null,那么这个node将是第一个节点
        enq(node);
        return node;
    }
    private Node enq(final Node node) {
            //一直循环直到设置成功
            for (;;) {
                Node t = tail;
                if (t == null) { // Must initialize
                    //必须初始化,直到设置成功为止
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    //当t不为null,说明初始化完成了,那么将当前节点node与之前的尾结点建立连接,并且替换尾结点
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }
    private final boolean compareAndSetTail(Node expect, Node update) {
        return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
    }
    

    主要流程如下:

    • 通过当前的线程和锁模式新创建一个节点
    • pred指针指向尾结点tail
    • 将new创建的Node的prev指向pred也就是之前的尾结点
    • 通过compareAndSetTail方法,完成尾结点的设置,这个方法主要对tailOffset和expect进行比较,如果tailOffset的Node和expect的Node地址相同的,那么设置tail的值为update的值。
    static {
            try {
                stateOffset = unsafe.objectFieldOffset
                    (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
                headOffset = unsafe.objectFieldOffset
                    (AbstractQueuedSynchronizer.class.getDeclaredField("head"));
                tailOffset = unsafe.objectFieldOffset
                    (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
                waitStatusOffset = unsafe.objectFieldOffset
                    (Node.class.getDeclaredField("waitStatus"));
                nextOffset = unsafe.objectFieldOffset
                    (Node.class.getDeclaredField("next"));
    
            } catch (Exception ex) { throw new Error(ex); }
        }
    

    从AQS的静态代码块可以看出,都是获取一个对象的属性相对于该对象在内存当中的偏移量,这样我们就可以根据这个偏移量在对象内存当中找到这个属性,tailOffext指的是tail对应的偏移量,所以这个时候会将nnew出来的Node置为当前队列的节点,同时由于是双向链表,也需要将前一个节点指向尾结点。

    • 如果pred指针是null,说明等待队列中没有元素,获取当前pred指针和tail指向的位置不同(说明被别的线程已经修改了),就需要看一下enq方法。
    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    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;
                }
            }
        }
    }
    

    如果没有被初始化,需要进行初始化一个头结点出来,但请注意,初始化的头结点并不是当前线程节点,而是调用了无参构造函数的节点。如果经历了初始化或者并发导致队列中有元素,则与之前方法相同,其实addWaiter就是一个双端链表添加尾结点的操作,需要注意的是,双端链表的头结点是一个无参的构造函数的头结点

    image

    总结一下,线程获取锁的时候,过程大致如下:

    • 当前没有线程获取到锁时,线程1获取锁成功
    • 线程2申请锁的时候,但是说被线程1占有
    • 如果再有线程要获取锁,一次在队列中往后排队即可。

    回到上边的代码,hasQueuedPredecessors()是公平锁加锁判断等待队列中是否存在有效节点,如果返回false,说明当前线程可以争取共享资源,如果返回true说明队列中存在有效节点,当前线程必须加入到等待队列中。

    public final boolean hasQueuedPredecessors() {
            Node t = tail; // Read fields in reverse initialization order
            Node h = head;
            Node s;
            return h != t &&
                ((s = h.next) == null || s.thread != Thread.currentThread());
        }
    

    看到这里我们理解一下h!=t&&((s=h.next)==null||s.thread!=Thread.currentThread());,为什么判断头结点的下一个节点?,第一个节点存储的数据是什么?
    双向链表中,第一个节点为虚节点,其实并不存储任何信息,只是占位,真正的第一个数据的节点,是在第二个节点开始的。当h!=t时:如果(s=h.next)==null,等待队列正在有线程进行初始化,但只是进行到了tail指向head,没有将head指向tail,此时队列中有元素,需要返回true(这块具体见下面代码),如果(s=h.next)!=null,说明此时队列中至少有一个有效节点,如果此时s.thread==Thread.currentThread,说明等待队列的第一个有效节点中的线程与当前线程相同,那么当前线程时可以获取资源的,如果s.thread!=Thread.currtThread(),说明等待队列第一个有效节点呢与当前线程不同,说明线程必须加入等待队列

    // java.util.concurrent.locks.AbstractQueuedSynchronizer#enq
    
    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;
                }
            }
        }
    }
    

    节点入队不是原子操作,所以会出现短暂的head!=tail,此时tail指向最后一个节点,而且tail指向head,如果head没有指向tail,这种情况下与需要相关线程加入队列中,所以这块代码是为了解决极端情况下的并发问题。

    2.3.1.3 等待队列中线程出列时机

    回到最初代码:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    public final void acquire(int arg) {
        if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    

    上文解释了addWaiter方法,这个方法其实就是把相应的线程以Node的数据结构加入到双端队列中,返回的是一个包含该线程的Node,而这个Node会作为参数,进入到acquireQueued方法中,acquireQueued方法可以对排队的线程进行或锁操作。

    总的来说,一个线程获取锁失败了,被放入等待队列中,acquireQueued会把放入队列中的线程不断去获取锁,直到获取成功或者不在需要获取(中断)。

    下面我们从何时出队列?和如何出队列,两个方向分析一下acquireQueued源码:

    final boolean acquireQueued(final Node node, int arg) {
            //标志是否成功拿到资源
            boolean failed = true;
            try {
                //标记等待过程中是否中断过
                boolean interrupted = false;
                //开始自旋,要么获取锁成功,要么中断
                for (;;) {
                    //获取当前节点的前驱节点
                    final Node p = node.predecessor();
                    //如果p是头结点,说明当前节点在真是的数据队列的的首部,就尝试获取锁(因为头结点是虚节点)
                    if (p == head && tryAcquire(arg)) {
                        //获取成功,头指针移动到当前的node
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    //说明p为头结点切当前没有获取到锁(可能是非公平锁抢占了)或者是p不为头结点,这个时候就要判断当前node是否要被阻塞(被阻塞条件:前驱节点waitStatus为-1),防止无限循环浪费资源,具体两个方法下面分析
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    

    注:setHead方法是把当前节点设置为虚节点,但并没有修改waitStatus,因为他是一直需要用的数据

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }
    
    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    // 靠前驱节点判断当前线程是否应该被阻塞
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 获取头结点的节点状态
        int ws = pred.waitStatus;
        // 说明头结点处于唤醒状态
        if (ws == Node.SIGNAL)
            return true; 
        // 通过枚举值我们知道waitStatus>0是取消状态
        if (ws > 0) {
            do {
                // 循环向前查找取消节点,把取消节点从队列中剔除
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 设置前任节点等待状态为SIGNAL
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    

    parkAndCheckInterrupt主要用于挂起当前线程,阻塞调用栈,返回当前线程的中断状态。

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }
    

    上述方法的流程如下:


    image

    从上图可以看出,跳出循环条件是当前置节点是头结点,且当前线程获取锁成功了。为了防止因死循环导致CPU资源被浪费,我们会判断前置节点的状态来决定是否将当前线程挂起,具体挂起流程用流程图表示如下(shouldParkAfterFailedAcquire的流程):

    image

    从队列中释放节点的疑虑取消了,那么又有新的问题了:

    • shouldParkAfterFailedAcquire中取消节点是怎么生成的,什么时候会把一个节点的waitSatus设置为-1呢?
    • 是在什么时间释放节点通知被挂起的线程呢?

    2.3.2 CANCELLED状态节点的生成

    acquireQueued方法中finally代码:

    final boolean acquireQueued(final Node node, int arg) {
            //标志是否成功拿到资源
            boolean failed = true;
            try {
                //标记等待过程中是否中断过
                boolean interrupted = false;
                //开始自旋,要么获取锁成功,要么中断
                for (;;) {
                    //获取当前节点的前驱节点
                    final Node p = node.predecessor();
                    //如果p是头结点,说明当前节点在真是的数据队列的的首部,就尝试获取锁(因为头结点是虚节点)
                    if (p == head && tryAcquire(arg)) {
                        //获取成功,头指针移动到当前的node
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    //说明p为头结点切当前没有获取到锁(可能是非公平锁抢占了)或者是p不为头结点,这个时候就要判断当前node是否要被阻塞(被阻塞条件:前驱节点waitStatus为-1),防止无限循环浪费资源,具体两个方法下面分析
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    

    通过cancelAcquire方法将Node状态标记为CANCELLED。接下来我们逐行分析这个方法原理:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    private void cancelAcquire(Node node) {
      // 将无效节点过滤
        if (node == null)
            return;
      // 设置该节点不关联任何线程,也就是虚节点
        node.thread = null;
        Node pred = node.prev;
      // 通过前驱节点,跳过取消状态的node
        while (pred.waitStatus > 0)
            node.prev = pred = pred.prev;
      // 获取过滤后的前驱节点的后继节点
        Node predNext = pred.next;
      // 把当前node的状态设置为CANCELLED
        node.waitStatus = Node.CANCELLED;
      // 如果当前节点是尾节点,将从后往前的第一个非取消状态的节点设置为尾节点
      // 更新失败的话,则进入else,如果更新成功,将tail的后继节点设置为null
        if (node == tail && compareAndSetTail(node, pred)) {
            compareAndSetNext(pred, predNext, null);
        } else {
            int ws;
        // 如果当前节点不是head的后继节点,1:判断当前节点前驱节点的是否为SIGNAL,2:如果不是,则把前驱节点设置为SINGAL看是否成功
        // 如果1和2中有一个为true,再判断当前节点的线程是否为null
        // 如果上述条件都满足,把当前节点的前驱节点的后继指针指向当前节点的后继节点
            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)
                    compareAndSetNext(pred, predNext, next);
            } else {
          // 如果当前节点是head的后继节点,或者上述条件不满足,那就唤醒当前节点的后继节点
                unparkSuccessor(node);
            }
            node.next = node; // help GC
        }
    }
    

    执行流程:

    • 获取当前节点的前驱节点,如果前驱节点是CANCELLED,那就一直往前遍历,知道找到一个waitStatus<=0的节点,将找到pred节点和当前node关联,将当前的node设置为CANCELLED.
    • 根据当前节点的位置,考虑以下三种情况。
      • 当前节点是尾结点
      • 当前节点是head的后继节点
      • 当前节点不是head的后继节点,也不是尾结点
        根据上述第二条,我们分析每一种情况的流程。

    当前节点是尾结点:


    image

    当前节点是head的后继节点:

    image
    当前节点不是head的后继节点也不是尾结点:
    image

    通过上面的流程我们对于CANCELLED节点状态产生和变化已经有了大致了解,但是为什么所以变化都是对next指针进行了操作,而没有对prev指针进行操做呢,什么时候会对prev指针进行操作?

    • 执行cancelAcquire的时候,当前节点的前置节点可能已经从队列中出去了(已经执行try代码块中的shouldParkAfterFailedAcquire方法了),如果此时修改prev指针,有可能会导致prev指向另一个已经移除队列的node,因此这块变化prev指针不安全,shouldParkAfterFailedAcquire方法中,会执行下面的代码,其实就是在处理prev指针,shouldParkAfterFailedAcquire是获取所失败的情况下才会执行,进入该方法后,说明共享资源已经被获取了,当前节点之前的节点都不会出现变化,因此这个时候变更prev指针是比较安全的。
    do {
        node.prev = pred = pred.prev;
    } while (pred.waitStatus > 0);
    

    2.3.3如何解锁

    上面已经剖析了加锁过程中的基本流程,接下来再对解锁进行分析,由于ReentrantLock在解锁时候并不区分公平锁和非公平锁,直接看解锁代码:

    // java.util.concurrent.locks.ReentrantLock
    
    public void unlock() {
        sync.release(1);
    }
    

    可以看到,本质释放锁的地方,是通过AQS框架来完成的:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    

    在ReentrantLock里面的公平锁和非公平锁的父类Sync定义了可重入锁的释放机制:

    // java.util.concurrent.locks.ReentrantLock.Sync
    
    // 方法返回当前锁是不是没有被线程持有
    protected final boolean tryRelease(int releases) {
        // 减少可重入次数
        int c = getState() - releases;
        // 当前线程不是持有锁的线程,抛出异常
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        // 如果持有线程全部释放,将当前独占锁所有线程设置为null,并更新state
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }
    

    我们来解释下面源码:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    public final boolean release(int arg) {
        // 上边自定义的tryRelease如果返回true,说明该锁没有被任何线程持有
        if (tryRelease(arg)) {
            // 获取头结点
            Node h = head;
            // 头结点不为空并且头结点的waitStatus不是初始化节点情况,解除线程挂起状态
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    

    这里的判断条件为什么是h!=null&&h.waitStatus!=0?

    • h==null head还没有初始化,初始情况下,head == null,第一个节点入队,head会被初始化一个虚拟节点,所以说这里如果还没来得及入队,就会出现head == null 的情况
    • h!=null&&waitStatus == 0 表明了后继节点对应的线程仍在运行,不需要唤醒
    • h!=null && waitStatus<0表明后继节点可能被阻塞了,需要唤醒

    在看一下unparkSuccessor方法:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    private void unparkSuccessor(Node node) {
        // 获取头结点waitStatus
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        // 获取当前节点的下一个节点
        Node s = node.next;
        // 如果下个节点是null或者下个节点被cancelled,就找到队列最开始的非cancelled的节点
        if (s == null || s.waitStatus > 0) {
            s = null;
            // 就从尾部节点开始找,到队首,找到队列第一个waitStatus<0的节点。
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        // 如果当前节点的下个节点不为空,而且状态<=0,就把当前节点unpark
        if (s != null)
            LockSupport.unpark(s.thread);
    }
    

    为啥你们你要从后往前找第一个非CANCELLED的节点呢,原因如下:之前的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;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }
    

    我们看到,节点入队并不是原子操作,也就是说node.prev=pred;compareAndSetTail(pred,node)这两个地方可以看做tail入队的原子操作,但是此时pred.next =node;还没有执行,如果这个时候执行unparkSuccessor方法,就没办法从前往后找了,所以需要从后往前找,还有一点原因,在产生CANCELLED状态节点时候,先断开的是next指针,prev指针并未断开因此也是必须要从后往前遍历才能遍历全部的node。

    综上所述,如果从前往后找,由于极端情况如归的非原子操作和CANCELLED节点产生过程中断开next指针操作,可能会导致无法遍历所有节点,所有唤醒对应的线程后,对应的线程就会继续执行,继续执行acquireQueued方法以后中断如何处理?

    2.3.4 中断恢复后的执行流程

    唤醒后,会执行return Thread.interrupted(),这个函数返回的是当前执行线程的中断状态,并清除。

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }
    

    再回到acquireQueued代码中,当parkAndCheckInterrupt返回true或者false的时候,interrupted的值不同,但都会执行下次循环,如果这个时候获取锁成功,就会把当前interrupted返回。

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    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; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                    interrupted = true;
                }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    

    如果acquireQueue为true,就会会执行selfInterrupt方法:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    static void selfInterrupt() {
        Thread.currentThread().interrupt();
    }
    

    该方法其实是为了中断线程,但为什么获取了锁,以后还要中断,线程呢?这部分属于Java提供的协作式中断知识内容:

    • 当中的线程被唤醒时,并不知道被唤醒原因,可能是当前线程在等待中被中断,也可能是释放了锁以后被唤醒。因此我们通过Thread.interrupted()方法检查中断标记(该方法返回了当前线程的中断状态,并将当前线程的中断标志设置为false),并记录下来,如果发现线程被中断过,就在中断一次。
    • 线程在等待资源过程中被唤醒,唤醒后还是会不断去尝试获取锁,直到抢到位置,也就是说,在整个流程中,并不响应中断,只是记录中断记录,最后抢到锁返回了,那么如果中断过的话,需要补充一次中断。

    这里的处理方式主要运用线程池基本运作单元World中的runWork,通过Thread.interrupted(),进行额外的处理。

    2.3.5 小结

    我们在1.3小节提出的一些问题,现在来回答

    • 某个线程获取锁失败的后续操作流程是什么?
      • 存在某种排队等候机制,线程继续等待,仍然保留获取锁的可能,获取锁流程仍在继续
    • 既然说到了排队等候机制,那么就一定会有某种队列形成,这样的队列是什么数据结构?
      • 是CLH变体的FIFO双端队列(双向链表实现的)
    • 处于排队等候机制过程,什么时候可以有机会获取锁?
      • 可以详细看2.3.1.3小结
    • 如果处于排队等候机制中线程一直无法获取锁,需要一直等待吗?还是有别的策略来解决这问题?
      • 线程所在节点的状态会变成取消状态,取消状态的节点会从队列中释放。具体可见2.3.2小结
    • lock函数通过acquire方法进行加锁,但是具体怎么实现的?
      • AQS的acquire会调用tryAcquire方法,tryAcquire由各个自定义同步器实现,通过tryAcquire完成加锁过程。

    3.AQS应用

    3.1 ReentrantLock的可重入应用

    ReentrantLock的可重入性是AQS很好的应用之一,在ReenTrantLock里面,不管公平锁还是非公平锁,都有一段逻辑:

    公平锁:

    // java.util.concurrent.locks.ReentrantLock.FairSync#tryAcquire
    
    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;
    }
    

    非公平锁:

    // java.util.concurrent.locks.ReentrantLock.Sync#nonfairTryAcquire
    
    if (c == 0) {
        if (compareAndSetState(0, acquires)){
            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;
    }
    

    上面的current == getExclusiveOwnerThread()就是判断当前线程是否是持有锁的线程,其中state来控制整体可重入的情况,state是volatile修饰的,用于保证一定的可见性。

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
    
    private volatile int state;
    

    接下来看state这个字段主要过程:

    • state初始化的时候为0,表示没有任何线程持有锁
    • 当有线程持有锁时,值就会在原来的基础上+1,同一个线程多次获的就会多次+1,这就是可重入的概念
    • 解锁也是对这个字段-1,一直到0,此线程对锁释放

    3.2 JUC中的应用场景

    除了上边的ReentrantLock的可重入性的应用,AQS作为并发的编程的框架,为很多同步工具提供了良好的解决方案,下面是AQS的应用场景:

    同步工具 同步工具与AQS的关联
    RenntrantLock 使用AQS保存锁重复持有的次数,当一个线程获取锁时,ReentLock记录当前获得锁的线程标识,用于检测是否重复获取,以及错误线程试图解锁操作时异常情况的处理。
    Semaphore 使用AQS同步状态来保存信号量的当前计数。tryRelease会增加计数,acquireShared会减少计数。
    CountDownLatch 使用AQS同步状态来表示计数,计数为0时,所有acquire(CountDownLatch的await方法)操作才可以通过。
    ReentrantReadWriteLock 使用AQS同步状态的高16位保存读锁的次数,低16位保存写锁次数
    ThreadPoolExecutor Worked利用AQS同步状态对独占线程变量的设置(tryAcquire和tryRelease)

    相关文章

      网友评论

          本文标题:从ReentrantLock的实现看AQS的原理及应用

          本文链接:https://www.haomeiwen.com/subject/mqgaahtx.html