AbstractQueuedSynchronizer与ReentrantLock

Posted 2020-11-24 insanexs

介绍

j.u.c包中的Lock定义了锁的行为。
而ReentrantLock是并发包下提供的一个锁的实现，它是一个可重入的、排他的锁。

ReentrantLock有的属性也很简单，除了一个serialVersionUID外，只有一个sync。
ReentrantLock可以分为公平锁和非公平锁两种。可以在创建时，通过向构造函数传入fair参数指定：

public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

无参的构造函数默认会创建一个非公平锁。公平锁和非公平锁的主要区别是：公平锁会保证在锁空闲时，先申请的线程先得到锁；而非公平锁则不考虑线程申请锁的先后顺序，随机让一个线程获得锁。
通过上文的构造函数我们也可以看到，其实内部负责锁的工作的是Sync。
ReentrantLock内部定义了一个抽象类Sync，定义了基本锁的行为，然后通过两个具体子类FairSync和NonfairSync调整具体实现细节，控制公平和非公平的实现（模版模式）。
Sync类继承自AbstractQueuedSynchronizer（AQS）。AQS是并发包中实现所有锁的基础（无论是共享锁/排他锁，可重入锁/不可重入锁）。其基本思想就是维持一个队列，队列中的每个节点保存了需要获取锁的线程，当线程无法获取锁时，会被Park从而进入WAITING状态，而当一个线程释放锁时，会从队列中选择一个节点的线程unPark。AQS通过CAS的操作和LockSupport类，在不用synchronize的情况下，实现了锁（因而效率也比synchronize要高）。
我们就通过对ReentrantLock上锁以及释放锁的过程进行分析，了解AQS。
在分析AQS前，同样对AQS内部拥有的变量做个分析。
AQS的成员变量主要可以分为三个类型（除序列化号外）：

• CAS操作相关：
  这部分主要是UNSAFE对象及AQS各属性的地址偏移量字段。
  

    private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final long stateOffset;
    private static final long headOffset;
    private static final long tailOffset;
    private static final long waitStatusOffset;
    private static final long nextOffset;

    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用state表示锁的状态，state = 0时，表示锁未被持有，而当state > 0时，表示锁被持有，且state表示被持有的次数

    private volatile int state;

AQS继承了AbstractOwnableSynchronzier，后者用一个字段表示当前持有锁的线程。可以通过该字段得到锁的持有者。

    private transient Thread exclusiveOwnerThread;

• AQS维护的队列相关
  AQS中定义了一个内部类Node，表示一个节点，节点构成双向链表，形成队列。
  

static final class Node {
    //节点内部有一个特殊的指针，指向特殊的Node，用来表示这个节点的模式，共享/排他
    static final Node SHARED = new Node();

    static final Node EXCLUSIVE = null;

    /*********节点的状态常量**********/ //取消（表示该节点等待锁的线程已经取消）    
    static final int CANCELLED =  1;
    //信号量 表示该节点需要负责后续节点的唤醒   
    static final int SIGNAL    = -1;

    static final int CONDITION = -2;
    
    static final int PROPAGATE = -3;
    
    //表示节点的状态
    volatile int waitStatus;
    
    //指向前继节点的指针
    volatile Node prev;

    //指向后继节点的指针
    volatile Node next;

    //节点代表的线程 
    volatile Thread thread;

    //特殊的指针，表示节点的模式
    Node nextWaiter;

    //是否是共享模式
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

    //获取前继节点 
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

    Node() {    
    }

    Node(Thread thread, Node mode) {
        this.nextWaiter = mode;
        this.thread = thread;
    }

    Node(Thread thread, int waitStatus) { 
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

AQS内部有两个字段保存链表的头节点和尾节点。

    private transient volatile Node head;

    private transient volatile Node tail;

ReentrantLock的上锁过程

阻塞lock()

lock()是一个阻塞的方法，线程会一直阻塞，直到成功获取锁。lock()方法内部实现很简单，将上锁的操作委托给sync,由sync.lock()方法实现。
sync.lock()方法对于FairSync和NonfairSync的实现是不一样的，本文只针对公平锁分析。非公平锁的流程留给读者自行学习。
FairSync.lock()内部是通过acquire(1)实现。

acquire()方法主要分为两步：

• 先调用tryAcquire()尝试获取锁
  java protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); //判断是否被持有 if (c == 0) { //由于是公平锁，即使未被持有，也有看队列中是否有在自己之前等待锁的线程 if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { //如果没有则用CAS更新state，并设置线程持有者 setExclusiveOwnerThread(current); return true; } } 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; }
  对于公平锁而言，tryAcquire()只会在以下几种情况返回true：
  1）线程本就是锁的持有者，此时线程只是重入锁，
  2）锁未被其他线程持有且没有其他线程排在该线程之前要获得锁。

• 如果尝试获取锁失败，则构建一个代表该线程的节点添加至队列，并将线程Park，等待前继节点释放锁时，唤醒该线程。
  

acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

因为ReentrantLock是排他的，所以在添加节点中，添加了一个排他的节点（模式为Node.EXCLUSIVE）。
添加进队列，并Park线程的操作主要在acquitrQueued中。

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;
                }
                
                //到这里说明获取锁失败
                //检查是否需要Park线程，如果需要则Park线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)            //唤醒失败，将这个节点的waitingStatus更新为CANCELLED
                cancelAcquire(node);
        }
    }

首先acquireQueued在外层套了一个自循环，可以让线程在获取锁失败后继续尝试以及避免线程的假唤醒。
第一个if分支，是判断该线程是否可以去获取锁，并是否获取成功，如果获取成功，则更新等候链表的头节点，并释放旧的头节点。
到第二个if分支时说明线程获取锁失败了，此时需要检查是否需要Park线程（此时一般说明该线程还没到实际去获取锁，因此Park掉线程，不让他一直尝试，浪费CPU时间）。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL) //前继节点为SIGNAL，说明此时线程还不需要去争抢锁
            return true;
        if (ws > 0) {//前继节点被取消，则更新任务链
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            //更新状态
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

返回false的情况后，线程会继续去尝试获取锁，如果再次失败依旧会进入到第二个分支并被决定是否要被Park。
线程会一直在acquireQueued()这个方法内阻塞，直到线程能够获取锁或是线程在上锁过程异常退出才会返回，lock方法才会执行结束。

ReentrantLock的释放锁过程

ReentrantLock.unlock()

Lock接口定义了多种上锁的方式，lock()(阻塞直到成功)，tryLock()(尝试上锁，无论结果直接返回)，tryLock(t)（在规定时间内尝试，超时未成功则返回失败），lockInteruptibly()（阻塞上锁，但是外部可中断）。然而对释放锁的过程就只定义了一种unlock()。
unlock()实现也比较简单，ReentrantLock同样把释放的动作委托给sync.release(1)：

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

Sync会调用tryRelease()尝试释放锁，如果释放成功，则考虑唤醒队列中的后继节点的线程。
对于trtRelease()而言，只要是排他锁（无论是公平还是非公平）流程都是一样的，因此这部分实现被提取到了抽象父类Sync中：

    protected final boolean tryRelease(int releases) {
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
        //因为已经判断了线程必须是锁的持有者，因此这里是线程安全的
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }

是否能释放，主要取决于操作的线程是否是锁的持有者，如果不是，会抛出IllegalMonitorStateException异常。如果是，则将锁的state值更新（减去释放的次数）。如果值已经更新为0，则说明锁已经被完全释放，清空锁的持有线程，最后返回释放结果。
如果释放成功，释放的线程还有责任要唤醒后继线程去获取锁。

    private void unparkSuccessor(Node node) {
        //更新node节点的值
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        //从链表尾部向前遍历，找到node节点的后继节点s
        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;
        }
        //如果s存在
        if (s != null)
            //唤醒后继节点的线程
            LockSupport.unpark(s.thread);
    }

以流程图的形式再回顾下lock和unlock过程

总结

如果将ReentrantLock的控制方式（公平锁为例）做个重口味的比方，就类似一群人排队去WC。当你跑到WC想要去方便时，看见门开着，而且没有其他人在排队（锁未被占有且AQS的队列没有其他线程在排队），你就可以自己先占好包厢，并且把门一关，这时候门上就提示有人（设置state和exclusiveOwnerThread）。其他人再想来只能在外边排起队来(addWaiter进AQS的队列)，而且其他人为了不浪费时间刷起了手机不在一直看WC有没有人（线程被Park起，让出CPU时间），期间你可能弄出了些动静，惊动了排队的人，他们会检查下是否轮到自己了，发现还没有后，依旧玩起了手机（被错误唤醒的线程继续被Park）。直到你方便完，你主动打开了包厢门，然后找到队列中第一个排队的人，告诉他说：哥们儿，轮到你了，请自便（tryRelease并且unparkSuccessor）。那个哥们就走进了厕所，关上了门。