乐趣区

从ReentrantLock到AQS的原理及应用

前言

Java 并发包(JUC:java.util.concurrent)中提供了很多并发工具,这其中,很多咱们耳熟能详的并发工具,ReentrantLock、Semaphore,它们的实现都用到了一个独特的基类 –AbstractQueuedSynchronizer, 简称 AQS。AQS 是一个用来构建锁和同步器的框架,应用 AQS 能简略且高效地结构出利用宽泛的大量的同步器,比方咱们提到的 ReentrantLock,Semaphore,其余的诸如 ReentrantReadWriteLock,SynchronousQueue,FutureTask 等等皆是基于 AQS 的。当然,咱们本人也能利用 AQS 十分轻松容易地结构出合乎咱们本人需要的同步器。

AQS 负责管理同步器类中的状态,他治理了一个整数状态信息,能够通过 getState、setState、compareAndSetState 等类型办法来进行操作。ReentrantLock 示意所有线程曾经反复获取所的次数,Semaphore 示意残余许可数量。

ReentrantLock

1.1 ReentrantLock 概览

ReentrantLock 是一个可重入的互斥锁,又被称为“独占锁”。ReentrantLock 类实现了 Lock,它领有与 synchronized 雷同的并发性和内存语义,然而增加了相似锁投票、定时锁等待和可中断锁等待的一些个性。此外,它还提供了在强烈争用状况下更佳的性能。(换句话说,当许多线程都想访问共享资源时,JVM 能够花更少的时候来调度线程,把更多工夫用在执行线程上。)
ReentrantLock 锁在同一个工夫点只能被一个线程锁持有。

可重入:ReentrantLock 锁,能够被单个线程屡次获取。

ReentrantLock 分为“偏心锁”和“非偏心锁”。它们的区别体现在获取锁的机制上是否偏心。ReentrantLock 在同一个工夫点只能被一个线程获取 (当某线程获取到“锁”时,其它线程就必须期待);ReentraantLock 是通过一个 FIFO(先进先出)的期待队列来治理获取该锁所有线程的。在“偏心锁”的机制下,线程顺次排队获取锁;而“非偏心锁”在锁是可获取状态时,不论本人是不是在队列的结尾都会尝试获取锁而不须要马上排队( 获取不到锁再排队)。

通常咱们会将 ReentrantLock 和 synchronized 做比拟,两者孰优孰劣,其实在 java 官网对 synchronized 进行了诸多优化之后(偏差锁、轻量锁、自适应自旋、锁打消、锁粗化 …),两者的性能差距并不大,只是在某些方面存在差别而已,在此用一张表格做一个简略的比照:

// **************************Synchronized 的应用形式 **************************
// 1. 用于代码块
synchronized (this) {}
// 2. 用于对象
synchronized (object) {}
// 3. 用于办法
public synchronized void test () {}
// 4. 可重入
for (int i = 0; i < 100; i++) {synchronized (this) {}}
// **************************ReentrantLock 的应用形式 **************************
public void test () throw Exception {
    // 1. 初始化抉择偏心锁、非偏心锁
    ReentrantLock lock = new ReentrantLock(true);
    // 2. 可用于代码块
    lock.lock();
    try {
        try {
            // 3. 反对多种加锁形式,比拟灵便; 具备可重入个性
            if(lock.tryLock(100, TimeUnit.MILLISECONDS)){}} finally {
            // 4. 手动开释锁
            lock.unlock()}
    } finally {lock.unlock();
    }
}

1.2 ReentrantLock 与 AQS

下面咱们对 ReentrantLock 有了一个大略的理解,接下来能够来看看 ReentrantLock 和 AQS 之间到底是什么关系呢,其实 ReentrantLock 的底层就是应用的 AQS 实现的,咱们一起来看看源码:

/**ReentrantLock 实现了 Lock 类 其中有一个字段,Sync **/
public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    
    private final Sync sync;

    /**
     * Sync 就是继承了 AQS 的抽象类,此锁的同步控制根底。* 上面将 Sync 细分为偏心锁和非偏心锁。* 应用 AQS 的 state 来示意锁的重入次数。*/
    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;

        /**
         * 形象的 lock 办法,别离给偏心锁和非偏心锁具体实现。*/
        abstract void lock();
        ...
    }

    /**
     * 非偏心锁
     */
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;
        ...
    }

    /**
     * 偏心锁
     */
    static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;
        
        final void lock() {acquire(1);
        }
        ...
    }

    /**
     * ReentrantLock 的默认构造方法,默认创立非偏心锁
     */
    public ReentrantLock() {sync = new NonfairSync();
    }

    /**
     * 通过传入 boolean 参数来决定创立偏心锁还是非偏心锁。*/
    public ReentrantLock(boolean fair) {sync = fair ? new FairSync() : new NonfairSync();}

ReentrantLock 中的偏心锁和非偏心锁实现形式差异不大,差异在于偏心锁判断是否间接进入队列,先看看非偏心锁的加锁流程源码:

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;
        /** 加锁办法 **/
        final void lock() {
            // 若通过 CAS 设置变量 State(同步状态)胜利,也就是获取锁胜利,则将以后线程设置为独占线程。if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
            // 若通过 CAS 设置变量 State(同步状态)失败,也就是获取锁失败,则进入 Acquire 办法进行后续解决。acquire(1);
        }

        protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);
        }
    }

再看下偏心锁源码中获锁的形式:

static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;

        final void lock() {acquire(1);
        }
        ...
    }

通过查看偏心锁和非偏心锁获取锁的源码得悉,都会通过 lock()办法区上锁,然而 lock()办法到底是怎么上锁的呢?带着这样的疑难,咱们持续跟踪一下会发现,这个办法属于 FairSync 和 NonfairSync 的父类 AQS(AbstractQueuedSynchronizer) 的外围办法。
在理解 AQS 之前,咱们带着几个疑难:

  • 下面提到的通过 CAS 尝试扭转 state 的状态来示意获取锁胜利或者失败,如果获取失败了,调用 acquire() 办法,是怎么实现的呢?
  • 获取不到锁,这个线程是在干嘛呢?一直尝试锁?还是挂起期待唤醒?

上面咱们会对 AQS 以及 ReentrantLock 和 AQS 的关联做具体介绍。

AQS

理解 AQS 之前,先看看 AQS 的整体框架:

  • 上图中有色彩的为 Method,无色彩的为 Attribution。
  • AQS 框架共分为五层,自上而下由浅入深,从 AQS 对外裸露的 API 到底层根底数据。
  • 当有自定义同步器接入时,只需重写第一层所须要的局部办法即可,不须要关注底层具体的实现流程。当自定义同步器进行加锁或者解锁操作时,先通过第一层的 API 进入 AQS 外部办法,而后通过第二层进行锁的获取,接着对于获取锁失败的流程,进入第三层和第四层的期待队列解决,而这些解决形式均依赖于第五层的根底数据提供层。

AQS 概述

AQS 保护一个用 Volatile 的 int 类型 state 的共享资源,通过内置的 FIFO 来实现获取资源线程的排队工作。( 这个内置的同步队列称为 ”CLH” 队列)。该队列由一个一个的 Node 结点组成,每个 Node 结点保护一个 prev 援用和 next 援用,别离指向本人的前驱和后继结点。AQS 保护两个指针,别离指向队列头部 head 和尾部 tail。

其实就是个变体 双端双向链表
当线程获取资源失败(比方 tryAcquire 时试图设置 state 状态失败),会被结构成一个结点退出 CLH 队列中,同时以后线程会被阻塞在队列中(通过 LockSupport.park 实现,其实是期待态)。当持有同步状态的线程开释同步状态时,会唤醒后继结点,而后此结点线程持续退出到对同步状态的抢夺中。

Node 节点

static final class Node {
        /** waitStatus 值,示意线程已被勾销(期待超时或者被中断)*/
        static final int CANCELLED =  1;
        /** waitStatus 值,示意后继线程须要被唤醒(unpaking)*/
        static final int SIGNAL    = -1;
        /**waitStatus 值,示意结点线程期待在 condition 上,当被 signal 后,会从期待队列转移到同步到队列中 */
        static final int CONDITION = -2;
       /** waitStatus 值,示意下一次共享式同步状态会被无条件地流传上来
        static final int PROPAGATE = -3;
        /** 期待状态,初始为 0 */
        volatile int waitStatus;
        /** 以后结点的前一个结点 */
        volatile Node prev;
        /** 以后结点的后一个结点 */
        volatile Node next;
        /** 与以后结点关联的排队中的线程 */
        volatile Thread thread;
        /** ...... */
    }

同步状态 State

/**
     * The synchronization state.
     */
    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);
    }

通过 AQS 中 state 的源码能够看出,对于 state 的办法都是 final 润饰的,阐明子类中无奈重写它们。咱们能够通过批改 State 字段示意的同步状态来实现多线程的 独占模式和共享模式(加锁过程)
一般来说,自定义同步器要么是独占形式,要么是共享形式,它们也只需实现 tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared 中的一种即可。AQS 也反对自定义同步器同时实现独占和共享两种形式,如 ReentrantReadWriteLock。ReentrantLock 是独占锁,所以实现了 tryAcquire-tryRelease。

独占模式(通过 ReentrantLock 了解 AQS)

获取同步状态 –acquire()

public final void acquire(int arg) {if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();}
  • tryAcquire: 首先,调用 tryAcquire 办法,若返回 true,意味着获取同步状态胜利,前面的逻辑不再执行;若返回 false,也就是获取同步状态失败,进入 acquireQueued 步骤;
  • acquireQueued: 此时,获取同步状态失败,结构独占式同步结点,通过 addWatiter 将此结点增加到同步队列的尾部(此时可能会有多个线程结点试图退出同步队列尾部,须要以线程平安的形式增加);
  • selfInterrupt: 该结点以在队列中尝试获取同步状态,若获取不到,则阻塞结点线程,直到被前驱结点唤醒或者被中断。

以非偏心锁为例,能够看到获取锁的大体流程

final boolean nonfairTryAcquire(int acquires) {
            // 获取以后线程
            final Thread current = Thread.currentThread();
            // 获取 state 的值
            int c = getState();
            // 如果 state 为 0,阐明以后没有其余线程占用共享资源,能够尝试获取锁
            if (c == 0) {
                // 用 CAS 批改 state 的值
                if (compareAndSetState(0, acquires)) {
                    // 批改胜利,获取锁胜利,解决以后线程
                    setExclusiveOwnerThread(current);
                    // 返回获取锁胜利
                    return true;
                }
            }
            // 如果 state 不为 0,判断占用锁的线程是否是以后线程
            else if (current == getExclusiveOwnerThread()) {
                // 如果是以后线程,对 state 做增量递增
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                    // 设置 state
                setState(nextc);
                // 返回获取锁胜利(可重入的原理)return true;
            }
            return false;
        }
  • 获取以后线程;
  • 获取 state 的值;
  • 如果 state 为 0,阐明以后没有其余线程占用共享资源,能够尝试获取锁;
  • 用 CAS 批改 state 的值;批改胜利,获取锁胜利,解决以后线程;
  • 返回获取锁胜利;
  • 如果 state 不为 0,判断占用锁的线程是否是以后线程;
  • 如果是以后线程,对 state 做增量递增;
  • 返回获取锁胜利(可重入的原理)。

值的留神得中央是,偏心锁中多了一部判断,再来看看偏心锁获取锁的大略流程:

protected final boolean tryAcquire(int acquires) {
            // 获取以后线程;final Thread current = Thread.currentThread();
            // 获取 state 的值;int c = getState();
            // 如果 state 为 0,阐明以后没有其余线程占用共享资源,能够尝试获取锁
            if (c == 0) {
                // 判断以后期待队列中是否存在无效节点的办法,// 如果返回 False,阐明以后线程能够争取共享资源;// 如果返回 True,阐明队列中存在无效节点,以后线程必须退出到期待队列中。// 用 CAS 批改 state 的值;批改胜利,获取锁胜利,解决以后线程;if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 如果 state 不为 0,判断占用锁的线程是否是以后线程;else if (current == getExclusiveOwnerThread()) {
                // 如果是以后线程,对 state 做增量递增;int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                    // 返回获取锁胜利(可重入的原理)。setState(nextc);
                return true;
            }
            return false;
        }
/** 判断以后期待队列中是否存在无效节点的办法 **/
public final boolean hasQueuedPredecessors() {
        // The correctness of this depends on head being initialized 其正确性取决于是否初始化头结点
        // before tail and on head.next being accurate if the current 如果以后节点在头结点和尾结点之间
        // thread is first in queue.  线程在队列中第一个
        // 获取以后尾结点
        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 时:1. 如果(s = h.next)==null,期待队列正在有线程进行初始化,但只是进行到了 Tail 指向 Head,没有将 Head 指向 Tail,此时队列中有元素,须要返回 True(这块具体见下边 addWaiter()曾经 enq()代码剖析)。2. 如果(s = h.next) != null,阐明此时队列中至多有一个无效节点,3. 如果此时 s.thread ==Thread.currentThread(),阐明期待队列的第一个无效节点中的线程与以后线程雷同,那么以后线程是能够获取资源的;4. 如果 s.thread != Thread.currentThread(),阐明期待队列的第一个无效节点线程与以后线程不同,以后线程必须退出进期待队列。

进入队列 addWaiter()

当尝试获取锁胜利,则间接完结并返回;
当尝试获取锁失败,则进入下一步: 线程退出期待队列addWaiter()

private Node addWaiter(Node mode) {
        // 用以后线程和传入的锁模式新建一个 Node
        Node node = new Node(Thread.currentThread(), mode);
        // 先获取到以后的尾结点,将尾结点指向到 Node pred
        Node pred = tail;
        // 如果获取到的尾结点不为空
        if (pred != null) {// 将新建节点的前置节点设置为下面获取到的 tail(tail 节点的后间接点此时为 null)
            node.prev = pred;
            // 利用 CAS 批改尾结点的值(将新建的 node 节点设置为 tail 尾结点)if (compareAndSetTail(pred, node)) {
                // 如果设置胜利了,将之前获取到的 pred(批改之前的尾结点)的后置节点批改为以后新建的 node 节点
                pred.next = node;
                // 返回批改胜利的新的节点(此时此新建的这个 node 节点为新的 tail 节点)return node;
            }
        }
        // 如果获取到的 tail 节点为 null,(阐明期待队列中没有元素),// 或者以后 Pred 指针和 Tail 指向的地位不同(阐明被别的线程曾经批改),//enq()中解决了以上两种状况
        enq(node);
        return node;
    }

如果没有被初始化,须要进行初始化一个头结点进去。但请留神,初始化的头结点并不是以后线程节点,而是调用了无参构造函数的节点。如果经验了初始化或者并发导致队列中有元素,则与之前的办法雷同。其实,addWaiter 就是一个在双端链表增加尾节点的操作,须要留神的是,双端链表的头结点是一个无参构造函数的头结点。

private Node enq(final Node node) {for (;;) {
            // 获取以后的尾结点指向到 t
            Node t = tail;
            // 如果此时尾结点为空
            if (t == null) { // Must initialize  必须要初始化一个头结点
                // 利用 CAS 初始化一个头部(调用无参结构,头肩点外部都为 null)if (compareAndSetHead(new Node()))
                    // 将尾结点设置为头节点(此时就一个节拍板结点和尾结点是一个)tail = head;
            } else {
                // 如果尾结点不为 null, 将新建的节点 node 的前置节点设置为 t
                node.prev = t;
                // 通过 CAS 批改尾结点,将尾结点指向为新的 node,
                // 这个办法次要是对 tailOffset 和 Expect 进行比拟,如果 tailOffset 的 Node 和 Expect 的 Node 地址是雷同的,// 那么设置 Tail 的值为 Update 的值。if (compareAndSetTail(t, node)) {// 批改胜利,将之前的尾结点(当初的倒数第二个节点)的后置节点设置为新的尾结点(node)
                    // 返回倒数第二个节点
                    t.next = node;
                    return t;
                }
            }
        }
    }

何时出列 acquireQueued()

通过下面的 addWaiter()办法咱们能够晓得,通过将一个新建的 Node 节点放入到队列里等待获取锁的资格,那么进入队列之后,怎么能力获取到锁呢?接下来咱们来看看线程怎么获取锁。
总的来说,一个线程获取锁失败了,被放入期待队列,acquireQueued 会把放入队列中的线程一直去获取锁,直到获取胜利或者不再须要获取(中断)。

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)) {
                    // 获取到锁,头指针挪动到以后 node
                    //setHead 办法是把以后节点置为虚节点,但并没有批改 waitStatus,因为它是始终须要用的数据。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);
        }
    }
/** 判断以后 node 是否要被阻塞 **/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 获取前置节点的阻塞状态
        int ws = pred.waitStatus;
        // 如果 Node.SIGNAL -1 状态,阐明前置节点的状态为唤醒状态
        if (ws == Node.SIGNAL)
            // 能够间接 park(间接阻塞)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;
    }

CANCELLED 状态节点生成 cancelAcquire()

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

以上从 ReentrantLock 的获取锁,上锁来对 AQS 的获取锁过程做了介绍,后续会持续将 AQS 开释锁的原理做介绍,未完待续 …

参考

  • 从 ReentrantLock 的实现看 AQS 的原理及利用
  • Java 并发包基石 -AQS 详解
退出移动版