• Java - ReentrantLock锁分析


    Java - JUC核心类AbstractQueuedSynchronizer(AQS)底层实现


     

     一.  AQS内部结构介绍

    JUC是Java中一个包   java.util.concurrent 。在这个包下,基本存放了Java中一些有关并发的类,包括并发工具,并发集合,锁等。

    AQS(抽象队列同步器)是JUC下的一个基础类,大多数的并发工具都是基于AQS实现的。

    AQS本质并没有实现太多的业务功能,只是对外提供了三点核心内容,来帮助实现其他的并发内容。

    三点核心内容:

    • int state
      • 比如ReentrantLock或者ReentrantReadWriteLock, 它们获取锁的方式,都是对state变量做修改实现的。
      • 比如CountDownLatch基于state作为计数器,同样的Semaphore也是用state记录资源个数。
    • Node对象组成的双向链表(AQS中)
      • 比如ReentrantLock,有一个线程没有拿到锁资源,当线程需要等待,则需要将线程封装为Node对象,将Node添加到双向链表,将线程挂起,等待即可。
    • Node对象组成的单向链表(AQS中的ConditionObject类中)
      • 比如ReentrantLock,一个线程持有锁资源时,执行了await方法(类比synchronized锁执行对象的wait方法),此时这个线程需要封装为Node对象,并添加到单向链表。

    二.  Lock锁和AQS关系

    ReentrantLock就是基于AQS实现的。ReentrantLock类中维护这个一个内部抽象类Sync,他继承了AQS类。ReentrantLock的lock和unlock方法就是调用的Sync的方法。

    AQS流程(简述)
    1. 当new了一个ReentrantLock时,AQS默认state值为0, head 和 tail 都为null;
    2. A线程执行lock方法,获取锁资源。
    3. A线程将state通过cas操作从0改为1,代表获取锁资源成功。
    4. B线程要获取锁资源时,锁资源被A线程持有。
    5. B线程获取锁资源失败,需要添加到双向链表中排队。
    6. 挂起B线程,等待A线程释放锁资源,再唤醒挂起的B线程。
    7. A线程释放锁资源,将state从1改为0,再唤醒head.next节点。
    8. B线程就可以重新尝试获取锁资源。
    注: 修改AQS双向链表时要保证一个私有属性变化和两个共有属性变化,只需要让tail变化保证原子性即可。不能先改tail(会破坏双向链表)

    三.  AQS - Lock锁的tryAcquire方法

    ReentrantLock中的lock方法实际是执行的Sync的lock方法。

    Sync是一个抽象类,继承了AQS

    Sync有两个子类实现:

    • FairSync: 公平锁
    • NonFairSync: 非公平锁

    Sync的lock方法实现:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    //  非公平锁
    final void lock() {
        //  CAS操作,尝试将state从0改为1
        //  成功就拿到锁资源, 失败执行acquire方法
        if (compareAndSetState(0, 1))
         // 成功就设置互斥锁的为当前线程拥有
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
     
    //  公平锁
    final void lock() {
        acquire(1);
    }

    如果CAS操作没有成功,需要执行acquire方法走后续

    acquire方法是AQS提供的,公平和非公平都是走的这个方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    public final void acquire(int arg) {
        //  1. tryAcquire方法: 再次尝试拿锁
        //  2. addWaiter方法: 没有获取到锁资源,去排队
        //  3. acquireQueued方法:挂起线程和后续被唤醒继续获取锁资源的逻辑
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
         // 如果这个过程中出现中断,在整个过程结束后再自我中断 
            selfInterrupt();
    }

    在AQS中tryAcquire是没有具体实现逻辑的,AQS直接在tryAcquire方法中抛出异常

    在公平锁和非公平锁中有自己的实现。

    • 非公平锁tryAcquire方法
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    //  非公平锁
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
     
    //  非公平锁再次尝试拿锁 (注:该方法属于Sync类中)
    final boolean nonfairTryAcquire(int acquires) {
        //  获取当前线程对象
        final Thread current = Thread.currentThread();
        //  获取state状态
        int c = getState();
        //  state是不是没有线程持有锁资源,可以尝试获取锁
        if (c == 0) {
            //  再次CAS操作尝试修改state状态从0改为1
            if (compareAndSetState(0, acquires)) {
                //  成功就设置互斥锁的为当前线程拥有
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        //  锁资源是否被当前线程所持有 (可重入锁)
        else if (current == getExclusiveOwnerThread()) {
            //  持有锁资源为当前, 则对state + 1
            int nextc = c + acquires;
            //  健壮性判断
            if (nextc < 0) // overflow
                //  超过最大锁重入次数会抛异常(几率很小,理论上存在)
                throw new Error("Maximum lock count exceeded");
            //  设置state状态,代表锁重入成功
            setState(nextc);
            return true;
        }
        return false;
    }
    • 公平锁tryAcquire方法
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    //  公平锁
    protected final boolean tryAcquire(int acquires) {
        //  获取当前线程对象
        final Thread current = Thread.currentThread();
        //  获取state状态
        int c = getState();
        //  state是不是没有线程持有锁资源
        if (c == 0) {
            //  当前锁资源没有被其他线程持有
            //  hasQueuedPredecessors方法: 锁资源没有被持有,进入队列排队
            //  排队规则:
            //  1. 检查队列没有线程排队,抢锁。
            //  2. 检查队列有线程排队,查看当前线程是否排在第一位,如果是抢锁,否则入队列(注:该方法只是判断,没有真正入队列)
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                //  再次CAS操作尝试, 成功就设置互斥锁的为当前线程拥有
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        //  锁资源是否被当前线程所持有 (可重入锁)
        else if (current == getExclusiveOwnerThread()) {
            //  持有锁资源为当前, 则对state + 1
            int nextc = c + acquires;
            //  健壮性判断
            if (nextc < 0)
                //  超过最大锁重入次数会抛异常(几率很小,理论上存在)
                throw new Error("Maximum lock count exceeded");
            //  设置state状态,代表锁重入成功
            setState(nextc);
            return true;
        }
        return false;
    }

    四.  AQS的addWaiter方法

     addWaiter方法,就是将当前线程封装为Node对象,并且插入到AQS的双向链表。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    //  线程入队列排队
    private Node addWaiter(Node mode) {
        //  将当前对象封装为Node对象 
        //  Node.EXCLUSIVE 表示互斥  Node.SHARED 表示共享
        Node node = new Node(Thread.currentThread(), mode);
        // 获取tail节点
        Node pred = tail;
        //  判断双向链表队列有没有初始化
        if (pred != null) {
            //  将当前线程封装的Node节点prev属性指向tail尾节点
            node.prev = pred;
            //  通过CAS操作设置当前线程封装的Node节点为尾节点
            if (compareAndSetTail(pred, node)) {
                //  成功则将上一个尾节点的next属性指向当前线程封装的Node节点
                pred.next = node;
                return node;
            }
        }
        //  没有初始化head 和 tail 都等于null
        //  enq方法: 插入双向链表和初始化双向链表
        enq(node);
        //  完成节点插入
        return node;
    }
     
    //  插入双向链表和初始化双向链表
    private Node enq(final Node node) {
        //  死循环 
        for (;;) {
            //  获取当前tail节点
            Node t = tail;
            //  判断尾节点是否初始
            if (t == null) { // Must initialize
                //  通过CAS操作初始化初始化一个虚拟的Node节点,赋给head节点
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                //  完成当前线程Node节点加入AQS双向链表的过程
                //  当前线程封装的Node的上一个prev属性指向tail节点
                //  流程: 1. prev(私有)  --->  2. tail(共有)  ---> 3. next (共有)
                node.prev = t;
                //  通过CAS操作修改tail尾节点指向当前线程封装的Node
                if (compareAndSetTail(t, node)) {
                    //  将当前线程封装的Node节点赋给上一个Node的下一个next属性
                    t.next = node;
                    return t;
                }
            }
        }
    }

    五.  AQS的acquireQueued方法

    acquireQueued方法主要就是线程挂起以及重新尝试获取锁资源的地方

    重新获取锁资源主要有两种情况:

    • 上来就排在head.next,就回去尝试拿锁
    • 唤醒之后尝试拿锁
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    //  当前线程Node添加到AQS队列后续操作
    final boolean acquireQueued(final Node node, int arg) {
        //  标记,记录拿锁状态   失败
        boolean failed = true;
        try {
            // 中断状态
            boolean interrupted = false;
            //  死循环
            for (;;) {
                //  获取当前节点的上一个节点    prev
                final Node p = node.predecessor();
                //  判断当前节点是否是head,是则代表当前节点排在第一位
                //  如果是第一位,执行tryAcquire方法尝试拿锁
                if (p == head && tryAcquire(arg)) {
                    //  都成功,代表拿到锁资源
                    //  将当前线程Node设置为head节点,同时将Node的thread 和 prev属性设置为null
                    setHead(node);
                    //  将上一个head的next属性设置为null,等待GC回收
                    p.next = null; // help GC
                    //  拿锁状态  成功
                    failed = false;
                    //  返回中断状态
                    return interrupted;
                }
                //  没有获取到锁 --- 尝试挂起线程
                //  shouldParkAfterFailedAcquire方法: 挂起线程前的准备
                //  parkAndCheckInterrupt方法: 挂起当前线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    //  设置中断线程状态
                    interrupted = true;
            }
        } finally {
            //  取消节点
            if (failed)
                cancelAcquire(node);
        }
    }
     
    //  检查并更新无法获取锁节点的状态
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        //  获取上一个节点的ws状态
        /**
        * SIGNAL(-1)   表示当前节点释放锁的时候,需要唤醒下一个节点。或者说后继节点在等待当前节点唤醒,后继节点入队时候,会将前驱节点更新给signal。
        * CANCELLED(1)  表示当前节点已取消调度。当timeout或者中断情况下,会触发变更为此状态,进入该状态后的节点不再变化。
        * CONDITION(-2)  当其他线程调用了condition的signal方法后,condition状态的节点会从等待队列转移到同步队列中,等待获取同步锁。
        * PROPAGATE(-3)   表示共享模式下,前驱节点不仅会唤醒其后继节点,同时也可能唤醒后继的后继节点。
        * 默认(0) 新节点入队时候的默认状态。
        */
        int ws = pred.waitStatus;
        //  判断上个节点ws状态是否是 -1, 是则挂起    
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) {
            /** 
            * 判断上个节点是否是取消或者其他状态。
            * 向前找到不是取消状态的节点,修改ws状态。
            * 注意:那些放弃的结点,由于被自己“加塞”到它们前边,它们相当于形成一个无引用链,
            * 稍后就会被GC回收,这个操作实际是把队列中的cancelled节点剔除掉。
            */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            //  如果前驱节点正常,那就把上一个节点的状态通过CAS的方式设置成-1
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
     
    //  挂起当前线程
    private final boolean parkAndCheckInterrupt() {
        //  挂起当前线程
        LockSupport.park(this);
        //  返回中断标志
        return Thread.interrupted();
    }

    六.  AQS的Lock锁的release方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    //  互斥锁模式   解锁
    public final boolean release(int arg) {
        //  尝试是否可以解锁
        if (tryRelease(arg)) {
            Node h = head;
            //  判断双链表是否存在线程排队
            if (h != null && h.waitStatus != 0)
                //  唤醒后续线程
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
     
    //  尝试是否可以解锁
    protected final boolean tryRelease(int releases) {
        //  锁状态 =  状态 - 1
        int c = getState() - releases;
        //  判断锁是是否是当前线程持有
        if (Thread.currentThread() != getExclusiveOwnerThread())
            //  当前线程没有持有抛出异常
            throw new IllegalMonitorStateException();
        boolean free = false;
        //  当前锁状态变为0,则清空锁归属线程
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        //  设置锁状态为0
        setState(c);
        return free;
    }
     
    //  唤醒线程
    private void unparkSuccessor(Node node) {
        //  获取头节点的状态
        int ws = node.waitStatus;
        if (ws < 0)
            //  通过CAS将头节点的状态设置为初始状态
            compareAndSetWaitStatus(node, ws, 0);
        //  后继节点
        Node s = node.next;
        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;
        }
        if (s != null)
            //  真正的唤醒操作
            LockSupport.unpark(s.thread);
    }

     

    以上仅供参考!!

  • 相关阅读:
    SpringMVC知识点总结2(孙哥)
    【学习笔记】Prufer序列
    C++数组空初始化器简介以及对比测试
    第2-1-2章 传统方式安装FastDFS-附FastDFS常用命令
    Virtio-PMD的路径选择与用法
    基于Python的QQ音乐音频图片搜索系统
    Rockwell EDI 850 采购订单报文详解
    图论算法<三>:Dijkstra算法介绍及分析
    [12] 使用 CUDA 加速排序算法
    基于SSM的医院住院管理系统的设计与实现
  • 原文地址:https://www.cnblogs.com/M-sakura/p/17602167.html