AbstractQueuedSynchronizer理解之四（Condition）

时间 2019-12-05

标签 abstractqueuedsynchronizer 理解之四 condition 繁體版

原文原文链接

什么是Condition

Condition必需要和独占锁一块儿使用，独占锁代替了原来的synchronized，Condition代替了原来的Object中的监视器方法（wait, notify and notifyAll）；一个Lock能够对应多个Condition，这样线程之间能够按照条件唤醒指定的线程，而不是简单的notifyAll多有的线程，使得咱们多线程编程的时候能够灵活的控制线程。java

独占锁和Condition最经典的配合使用就是ArrayBlockingQueue.java，典型的生产者消费者问题：node

/*
 * Concurrency control uses the classic two-condition algorithm
 * found in any textbook.
 */

/** Main lock guarding all access */
final ReentrantLock lock;

/** Condition for waiting takes */
private final Condition notEmpty;

/** Condition for waiting puts */
private final Condition notFull;

这是在许多教科书中能找到的经典的双Condition算法的并发控制，须要有一个独占锁ReentrantLock，而后再定义两个Condition，notEmpty（队列不是空的）表示能够从队列中消费元素的信号条件，notFull（队列不是满的）表示能够向队列生产元素的信号条件。这两个Condition都是调用了lock.newCondition()方法实例化的。算法

当消费者线程调用消费方法take时：编程

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        //当队列的元素数量为0时，调用notEmpty.await，阻塞当前的消费线程
        while (count == 0)
            notEmpty.await();
        //dequeue中调用了notFull.signal()，通知生产者队列还没满，能够生产
        return dequeue();
    } finally {
        lock.unlock();
    }
}

当生产者线程调用生产方法put时：安全

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        //当队列满时，调用notFull.await()，阻塞当前生产线程，中止生产
        while (count == items.length)
            notFull.await();
        //enqueue中调用了notEmpty.signal()，通知消费者队列里有元素，能够消费
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

Condition的await

在AQS中有一个ConditionObject内部类实现了Condition接口，其中有两个成员变量：多线程

/** First node of condition queue. */
    private transient Node firstWaiter;
    /** Last node of condition queue. */
    private transient Node lastWaiter;

Condition也有一个node队列，firstWaiter、lastWaiter分别表示第一个和最后一个node。并发

先看await方法：less

public final void await() throws InterruptedException {
        //若是线程设置中断标志，抛出中断异常
        if (Thread.interrupted())
            throw new InterruptedException();
        //往队列添加node
        Node node = addConditionWaiter();
        //彻底释放锁，head的后继节点将被唤醒，而后被移出sync队列
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        //判断当前节点是否在sync队列中（当condition调用signal是会将该节点放入Sync队列），若是不在就park当前线程，线程在这里开始等待被signal
        while (!isOnSyncQueue(node)) {
            LockSupport.park(this);
            //发送中断时（唤醒了线程）break；checkInterruptWhileWaiting中调用了transferAfterCancelledWait（贴在下面），这个方法时检测中断是发生在signal以前仍是以后
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        //当前线程被signal后，调用acquireQueued抢占锁，若是interruptMode不为抛出异常，设置为REINTERRUPT
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            //从头至尾移除取消的节点
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            //继续中断仍是抛出异常
            reportInterruptAfterWait(interruptMode);
    }
        
final boolean transferAfterCancelledWait(Node node) {
    //首先CAS设置node状态为0，若是成功说明中断发生在signal以前（由于signal会将node状态设置为0）
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        //将node入sync队列
        enq(node);
        return true;
    }
    /*
     * If we lost out to a signal(), then we can't proceed
     * until it finishes its enq().  Cancelling during an
     * incomplete transfer is both rare and transient, so just
     * spin.
     */
    //若是node不在sync队列中，yield，让出cpu
    while (!isOnSyncQueue(node))
        Thread.yield();
    //中断发生在signal后
    return false;
}

分析一下addConditionWaiter：oop

private Node addConditionWaiter() {
        Node t = lastWaiter;
        // If lastWaiter is cancelled, clean out.
        //若是最后一个node被取消，清除node
        if (t != null && t.waitStatus != Node.CONDITION) {
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        //新建一个node，持有当前线程，状态为CONDITION
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null)
            //若是尾节点为null，说明condition队列仍是空的，将新建的node做为头节点
            firstWaiter = node;
        else
            //若是condition队列已经存在，将新建的node做为尾节点的next
            t.nextWaiter = node;
        //将新建node设置为尾节点
        lastWaiter = node;
        //返回新建的node
        return node;
    }

在这里咱们能够看到Condition的队列是一个单链表。
看一下unlinkCancelledWaiters，Condition全部操做都是在获取锁以后执行的，因此不用考虑线程安全问题：ui

private void unlinkCancelledWaiters() {
        Node t = firstWaiter;
        Node trail = null;
        while (t != null) {
            Node next = t.nextWaiter;
            if (t.waitStatus != Node.CONDITION) {
                t.nextWaiter = null;
                if (trail == null)
                    firstWaiter = next;
                else
                    trail.nextWaiter = next;
                if (next == null)
                    lastWaiter = trail;
            }
            else
                trail = t;
            t = next;
        }
    }

该方法从队列头开始日后遍历全部node，移除已经取消的node；

在新建了node后，调用了fullyRelease：

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        //保存当前的state
        int savedState = getState();
        //release(savedState)尝试释放锁，这也是为何叫fullyRelease
        if (release(savedState)) {
            failed = false;
            //返回以前保存的state
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            //若是失败，将当前node设置为取消状态
            node.waitStatus = Node.CANCELLED;
    }
}

看一下release：

public final boolean release(int arg) {
    //尝试释放锁，这里调用的是ReentrantLock实现的tryRelease，传入的arg是当前的state，因此会释放成功，即state为0
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            //唤醒后继节点
            unparkSuccessor(h);
        return true;
    }
    return false;
}

下面的方法是判断当前节点是否在Sync队列中

final boolean isOnSyncQueue(Node node) {
    //若是当前节点状态为CONDITION或者节点前驱为null，说明该节点已经在CONDITION队列中，不在Syc队列里
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    //若是节点后继不是null，那该节点必定在Syc队列中
    if (node.next != null) // If has successor, it must be on queue
        return true;
    /*
     * node.prev can be non-null, but not yet on queue because
     * the CAS to place it on queue can fail. So we have to
     * traverse from tail to make sure it actually made it.  It
     * will always be near the tail in calls to this method, and
     * unless the CAS failed (which is unlikely), it will be
     * there, so we hardly ever traverse much.
     */
    //此时节点入列的CAS动做可能失败，因此要从尾部往前查找该节点再次确认
    return findNodeFromTail(node);
}

Condition的signal

public final void signal() {
        //若是当前线程不是当前的独占线程，抛出异常
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
            //signal Condition队列的第一个节点
            doSignal(first);
    }
    
    private void doSignal(Node first) {
        //若是transferForSignal失败（即当前节点取消）且下一个节点存在，while继续loop
        do {
            //设置第一个节点的next为firstWaiter，此时若是firstWaiter为null，说明队列空了，将lastWaiter也设置为null
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            //设置第一个节点next为null，help GC
            first.nextWaiter = null;
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }
    
    final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
     //若是为node设置状态失败，说明node被取消，返回false
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    /*
     * Splice onto queue and try to set waitStatus of predecessor to
     * indicate that thread is (probably) waiting. If cancelled or
     * attempt to set waitStatus fails, wake up to resync (in which
     * case the waitStatus can be transiently and harmlessly wrong).
     */
    //将当前node入列sync队列，返回node的前继
    Node p = enq(node);
    int ws = p.waitStatus;
    //若是前继的状态为取消或者设置前继状态为SIGNAL失败，当前node线程unpark
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

signal后，Condition第一个节点将入列sync的队列，等待抢占到锁继续执行。

总结

在一开是的例子中，假设有两个线程P，C分别表明生产者和消费者线程，生产消费元素E的队列Q容量为1。

C无限loop调用take，当C抢占到独占锁，发现Q时空的，调用notEmpty.await()，线程C释放锁而且入列notEmpty队列park，等待别的线程调用notEmpty.signal()；

P无限loop调用put，当P抢占到独占锁生产了一个E，调用notEmpty.signal()通知C，而后释放了锁；

C收到signal信号，入列SYC队列，而且unpark，尝试抢占独占锁，成功得到独占锁后，消费了一个E，而后调用notFull.signal()；

P生产E时发现Q已满（C还没来得及消费），调用notFull.await()线程P释放锁而且入列notFull队列park，等待notFull.signal()通知本身unpark并入列AQS队列去抢占独占锁进行生产；