Java并发（十一）：Condition条件

2018-11-20 03:21:05来源：博客园阅读 ()

synchronized配合Object的wait()、notify()系列方法可以实现等待/通知模式。

Lock提供了条件Condition，对线程的等待、唤醒操作更加详细和灵活。

Condition的作用是对锁进行更精确的控制。Condition中的await()方法相当于Object的wait()方法，Condition中的signal()方法相当于Object的notify()方法，Condition中的signalAll()相当于Object的notifyAll()方法。

一、Condition实现生产者消费者问题

class BoundedBuffer {
    final Lock lock = new ReentrantLock();
    // condition 依赖于 lock 来产生
    final Condition notFull = lock.newCondition();
    final Condition notEmpty = lock.newCondition();

    final Object[] items = new Object[100];
    int putptr, takeptr, count;

    // 生产
    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length)
                notFull.await();  // 队列已满，等待，直到 not full 才能继续生产
            items[putptr] = x;
            if (++putptr == items.length) putptr = 0;
            ++count;
            notEmpty.signal(); // 生产成功，队列已经 not empty 了，发个通知出去
        } finally {
            lock.unlock();
        }
    }

    // 消费
    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0)
                notEmpty.await(); // 队列为空，等待，直到队列 not empty，才能继续消费
            Object x = items[takeptr];
            if (++takeptr == items.length) takeptr = 0;
            --count;
            notFull.signal(); // 被我消费掉一个，队列 not full 了，发个通知出去
            return x;
        } finally {
            lock.unlock();
        }
    }
}

二、lock与condition关系

    // 一个lock可以new多个条件
    public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        lock.lock();
        Condition newCondition1 = lock.newCondition();
        newCondition1.await();
        newCondition1.signal();
        Condition newCondition2 = lock.newCondition();
        newCondition2.await();
        newCondition2.signal();
        lock.unlock();
    }

    // ReentrantLock.newCondition()
    public Condition newCondition() {
        return sync.newCondition();
    }
    
    // ReentrantLock.Sync.newCondition()
    final ConditionObject newCondition() {
        return new ConditionObject();
    }

三、条件队列

每个Condition对象都包含着一个FIFO队列，该队列是Condition对象通知/等待功能的关键。在队列中每一个节点（使用的AQS的节点）都包含着一个线程引用，该线程就是在该Condition对象上等待的线程。

public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    private transient Node firstWaiter; //头节点
    private transient Node lastWaiter; //尾节点
    public ConditionObject() {
    }
    
    /**
     * 通过addConditionWaiter()方法理解等待队列数据结构
     * 将当前线程加入条件等待队列
     * 1.Node就是AQS的Node
     * 2.单向链表，通过nextWaiter连接
     * 3.waitStatus==Node.CONDITION才能在等待队列中
     */
    private Node addConditionWaiter() {
        Node t = lastWaiter;
        if (t != null && t.waitStatus != Node.CONDITION) {
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null)
            firstWaiter = node;
        else
            t.nextWaiter = node;
        lastWaiter = node;
        return node;
    }
    
    /**
     * 清除队列中不是等待状态的线程
     */
    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;
        }
    }
}

四、等待await()

    public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        Node node = addConditionWaiter(); // 当前线程new Node()加入条件队列
        int savedState = fullyRelease(node); // 释放当先线程的锁
        int interruptMode = 0;
        
        /**
         * 自旋：
         * 1.当前节点不在同步队列(刚new的节点肯定不在)，挂起当前线程，等待被唤醒
         * 2.当其他线程调用同一个ConditionObject的signal方法时，会将队列里的节点放入同步队列，并unpark线程（排队唤醒）
         * 3.如果该节点被唤醒，再自旋检查是否在同步队列。发现已经在队列中，就可以跳出循环，获取lock
         */
        while (!isOnSyncQueue(node)) {
            LockSupport.park(this);
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) // 处理打断
                break;
        }
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE) // 获取锁
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

五、唤醒signal()

public final void signal() {
        if (!isHeldExclusively()) // 检查是否拿到锁
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
            doSignal(first);
    }
    
    private void doSignal(Node first) {
        do {
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            first.nextWaiter = null;
        } while (!transferForSignal(first) && // 唤醒队列第一个节点
                 (first = firstWaiter) != null);
    }
    
    final boolean transferForSignal(Node node) {
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        Node p = enq(node); // 加入同步队列
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread); // 唤醒线程
        return true;
    }