AbstractQueuedSynchronizer原理剖析

时间 2019-12-09

标签 abstractqueuedsynchronizer 原理剖析繁體版

原文原文链接

不管是公平锁仍是非公平锁，它们的实现都依赖于AbstractQueuedSynchronizer，它提供了一个基于先进先出等待队列实现block locks和synchronizers的框架。特性以下node

仅经过一个 int 类型来表明状态。对于ReentrantLock而言，他就是线程持有锁的次数，当次数为0时，表明锁没有被持有，正数表明被持有的次数，负数则是超出了锁的持有范围，有可能存在死循环
支持独占模式(默认的)和共享模式。在独占模式中，去获取一个已经被其它线程拥有的锁只会失败，共享模式中，则多个线程是能够成功的。

lock()原理

当ReentrantLock获取锁失败时，会执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)框架

private Node addWaiter(Node mode) { 
    Node node = new Node(Thread.currentThread(), mode); //建立一个节点，存储当前的线程，以及锁持有的模式，对于 ReentrantLock来讲就是 独占 型
    // 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)) {//CAS操做，若是当前的尾部节点没有被其它线程更改，那么把新的节点设置成队列的尾部
            pred.next = node;
            return node;
        }
    }
    enq(node);//首次入队
    return node;
}

获取失败进行入队操做，首先就是往队列中添加一个正在等待的节点Node性能

从Node自己的结构能够看到，AQS(AbstractQueuedSynchronizer)自己就维护了一个双向链表，用来存放等待中的线程。链表的每一个节点，表明那个线程，是独占仍是共享锁。
建立好节点以后，便执行入队操做，对于首次建立队列ui

private Node enq(final Node node) {
    for (;;) {
       //借助CAS机制实现无锁操做，因此须要一直执行直到CAS成功
        Node t = tail;
        if (t == null) { // 初始化发生在第一次建立队列，这样的好处是，当竞争不激烈的时候，实际上也就不会发生这些操做，性能也会好些
             if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

能够看到，入队也就是从队尾插入新的等待线程，入队完毕，也就开始去进行不断的尝试，直到获取锁成功,能够看到，对于lock来讲，其实已是阻塞了this

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)) { //仅当当前节点的前一个节点是head,才去获取线程，这里能够看出其实先等待的线程是会优先处理，也就是FIFO原则
                setHead(node); 
                p.next = null; // help GC    ,释放掉当前线程在队列中的引用，也能够看作’出队'了            
                failed = false;
                //执行到这里说明获取锁成功
                return interrupted;
            }
            //执行到这里说明存在竞争，有多个线程都在等待一个锁
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt()) //里面会对当前线程执行中断，当被唤醒时，继续循环
                    //若是线程被中断，设置中断标记，区别于 doAcquireInterruptibly，doAcquireInterruptibly是直接抛出异常，这也就是 lockInterruptibly可以抛出中断的缘由
                    interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

从这里能够看到，不管锁是公平锁仍是非公平锁，只要被放入了等待队列，此时的执行依然是谁先等待就先执行谁 ,非公平锁体如今新来的线程会无视已经等了的线程，能够优先去抢锁,因此公平体如今第一次参与抢锁的线程会去等待已经在等待队列中的线程,非公平并非说从已经在等待的线程队列里面随便选一个

shouldParkAfterFailedAcquire的源码以下spa

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus; //查看前一个节点的等待状态
    if (ws == Node.SIGNAL)
            //已经尝试过获取锁，能够执行park了
            return true;
    if (ws > 0) {   
        do {
            //去掉队列中全部已经取消的线程
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {             
        //此时当前线程的前一个节点的等待状态一定是0或者PROGATE，这代表当前线程在park以前能够再尝试一次去获取锁,也就是说前一个节点可能刚获取到SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

waitStatus：等待的状态，共有5种线程

SIGNAL:，代表它的前一个节点须要执行 unparking；
CANCELLED：当前节点保存的线程因为超时或者中断被取消了；
CONDITION：接档正处于条件队列中，执行了await；
PROPAGATE：一个共享的锁须要传递释放信号到其它节点
0:非上述4中状态，有多是刚获取signal,此时它的值是0，也有多是新建的head节点

parkAndCheckInterrupt主要是park当前线程code

private final boolean parkAndCheckInterrupt() {
    //当获取不到许可时，阻塞线程，解除阻塞状态的状况以下：
    //1 某个线程对这个线程调用了unpark方法
    //2 某个线程中断了这个线程
    //3 这个方法毫无理由的返回了 [park比较奇特的地方],基于这样，调用的时候必须去判断park的条件，以及当它返回的时候，去设置中断的状态
    LockSupport.park(this); 
    //返回线程的中断状态
    return Thread.interrupted();
}

至此lock()执行结束队列

unlock()原理

当执行unlock时，ReentrentLock执行对应的Release图片

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        //执行这里表示所已经被释放,可让它的下一个节点来抢锁
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h); //h.waitStatus == 0 表示尚未执行park,天然不须要unpark
        return true;
    }
    return false;
}

若是release成功，即当前线程持有的全部锁都已经释放，那么就能够执行 unparkSuccessor,从源码能够看到，unpark是从头部开始进行的，结合lock的原理，可知AQS自己就是一个先进先出的队列
unparkSuccessor源码以下

private void unparkSuccessor(Node node) {
     int ws = node.waitStatus;
    if (ws < 0)
        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); //恢复线程
}

至此unlock()完毕

await的原理

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();//当前线程已经中断了，抛出中断异常
    //添加一个新的waiter到condition queue中,这个新的Node的waitStatus会被标记为CONDITION
    Node node = addConditionWaiter(); 
    //释放当前线程拥有的锁,即从sync queue中去掉当前线程
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
    //若是当前线程不在持有锁的队列里头,对他进行休眠，当其它线程执行 unlock的时候，释放锁，就会执行unpark操做，此时它会被唤醒，唤醒后，若是它在syn队列里头，开始继续往下执行。(这个插入操做则是由signal完成)
        LockSupport.park(this); 
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;//等待的过程当中线程中断了，退出
    }
//从新竞争锁,至关于执行了lock操做
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
    //再次去获取锁，若是当前的线程在park的时候是被中断了，而且ConditionObject并非因为中断返回，这里再次标记为中断
        interruptMode = REINTERRUPT;  
    if (node.nextWaiter != null) 
    //清除非Condition模式的线程,而在signal中有先关操做将conditon的线程设置成非condition
         unlinkCancelledWaiters();
    if (interruptMode != 0)
    //上报等待的过程当中发生了中断,若是是要抛出中断，就抛出，不然再次执行中断
        reportInterruptAfterWait(interruptMode); 
}

isOnSyncQueue源码以下

final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)
    //node自己是调用了 await 方法,或者没有在获取锁的队列里头,[若是在里头一定有一个前置的节点]
        return false; 
    if (node.next != null) 
    //当前节点存在下一个节点，那么它确定是执行过 enq ,即获取过锁
        return true; 
    // CAS失败的时候，有可能 node.rev是没有的，所以须要从头至尾遍历一次
   return findNodeFromTail(node); 
}

checkInterruptWhileWaiting源码以下

private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}
final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
    //线程中断从新获取锁,而且设置waitStatus为0，以便后续线程从condition queue清除
        enq(node); 
        return true;
    }
    while (!isOnSyncQueue(node))
    //若是CAS失败，只要当前节点没有在Sync queue中，那么一直自旋，每次都会交出执行权限
        Thread.yield(); 
    return false;
}

能够看到，await其实就是释放线程原有的锁，并把它放入conditon队列中，而后执行阻塞。等唤醒的时候，从新获取锁，并清掉condition queue中的线程。至此await执行结束

singnal的原理

public final void signal() {
    if (!isHeldExclusively()) //只有当前线程持有了锁，才能释放
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);//优先释放队列头的,也就是等待时间最长的condition node
}
private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}
//将节点从condition queue转移到sync queue
final boolean transferForSignal(Node node) {
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false; //设置为非等待失败，则不继续转移
//CAS设置等待状态为0成功
Node p = enq(node); //新节点放入sync queue，并返回原来的尾部节点，也就是新节点的前一个节点
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) //参考shouldParkAfterFailedAcquire
        LockSupport.unpark(node.thread);//若是当前节点的前一个节点线程已经取消,或者将当前节点的前一个节点线程的waitStatus设置成SIGNAL失败,则直接唤醒当前线程
    return true;
}

能够看到signal最关键的信息就是去掉等待队列中的CONDITION状态，并将线程加入sync队列，至此signal结束