java并发编程之wait&notify VS lock&condition

jdk5之前线程同步可以用synchronized/wait/notify来进行控制，jdk5以后新添加了lock/condition。他们之间有什么联系与区别的？本文就用一个例子循序渐进的给大家展示一下：

首先来看一个有界缓存的例子：

abstract class BaseBoundedBuffer<V> {
    private final V[] buff;
    private int tail;
    private int head;
    private int count;
    protected BaseBoundedBuffer(int capacity){
        this.buff = (V[])new Object[capacity];
    }
    protected synchronized final void doPut(V v){//存
        buff[tail] = v;
        tail++;
        if(tail == buff.length){
            tail = 0;
        }
        count++;
    }
    protected synchronized final V doTake(){//get
        V v = buff[head];
        buff[head] = null;
        head++;
        if(head == buff.length){
            head = 0;
        }
        count--;
        return v;
    }
    protected synchronized final boolean isFull(){//是否是满的
        return count == buff.length;
    }
    protected synchronized final boolean isEmpty(){//是否是空的
        return count == 0;
    }
}

class GrumpBoundedBufer<V> extends BaseBoundedBuffer<V>{
    public GrumpBoundedBufer(int size){
        super(size);
    }
    public synchronized void put(V v)throws BufferFullException{
        if(isFull()){//存的时候，如果是满的，就抛异常
            throw new BufferFullException();
        }
        doPut(v);
    }
    public synchronized V take()throws BufferEmptyException{
        if(isEmpty()){//取的时候，如果是空的，就抛异常
            throw new BufferEmptyException();
        }
        return doTake();
    }
}

当然，上面的这种实现非常不友好，如果不满足先验条件就抛出异常，但是在多线程条件下，先验条件不会保持一个一成不变的状态，队列里面的元素是在不停的变化的，因此我们用轮询加休眠改进一下：

class SleepyBoundedBufer<V> extends BaseBoundedBuffer<V>{
    public SleepyBoundedBufer(int size){
        super(size);
    }
    public void put(V v) throws InterruptedException {
        while(true){
            synchronized(this){
                if(!isFull()){//如果不是满的，可以存
                    doPut(v);
                    return;
                }
            }
            //如果是满的，休眠1秒钟，然后重试
            Thread.sleep(1000);
        }
    }
    public V take() throws InterruptedException {
        while(true){
            synchronized(this){
                if(!isEmpty()){//如果不是空的，就可以取
                    return doTake();
                }
            }
            //如果是空的，休眠1秒钟，重试
            Thread.sleep(1000);
        }
    }
}

这种轮训+休眠的方式的缺点：
（1）休眠多少时间合适呢？
（2）给调用者提出处理InterruptedException的新的要求，因为sleep是会抛出这个异常的。

如果存在一种线程挂起的方式，它能保证，在某个条件变为真的时候，线程可以及时的苏醒过来，那就太好了！这就是条件队列所做的事情。

使用内部条件队列的实现方式：

class BoundedBufer<V> extends BaseBoundedBuffer<V>{
    protected BoundedBufer(int size) {
        super(size);
    }
    public synchronized void put(V v) throws InterruptedException {
        while(isFull()){//注意这里的while，而不是if
            wait();//如果是满的，把当前线程挂起
        }
        doPut(v);//如果不满，就可以存
        notifyAll();//存了以后，唤醒所有的等待线程，因为可能有线程在等待取，放进来以后就可以取了
    }
    public synchronized V take() throws InterruptedException {
        while(isEmpty()){//注意这里的while，而不是if
            wait();//如果是空的，把当前线程挂起
        }
        V v = doTake();//如果不空，取出来
        notifyAll();//然后唤醒所有的等待线程，因为有的线程可能在等待放，取出来以后就可以放了
        return v;
    }
}

这也是jdk5之前的解决方式。
条件队列可以让一组线程（叫做：等待集wait set）以某种方式等待相关条件变为真，条件队列的元素不同于一般的队列，一般队列的元素是数据项，条件队列的元素是线程。每个java对象都有一个内部锁，同时还有一个内部条件队列。一个对象的内部锁和内部条件队列是关联在一块的。Object.wait会自动释放锁，并请求os挂起当前线程，这样就给其他线程获得锁并修改对象状态的机会，当线程被唤醒以后，它会重新去获取锁。调用wait以后，线程就进入了对象的内部条件队列里面等待，调用notify以后，就从对象的内部条件队列里面选择一个等待线程，唤醒。 因为会有多个线程因为不同的原因在同一个条件队列中等待，因此，用notify而不用notifyAll是危险的！有的线程是在take()的时候阻塞，它等待的条件是队列不空，有的线程是在put()的时候阻塞，它等待的条件是队列非满。 如果调用了take()以后notify的是总是阻塞在take上的线程，就挂了！

BoundedBufer的put和take是一种很保守的做法，每次向队列里面添加或者移除都进行notifyAll，可以进行如下的优化：
是有从空变为了非空，或者是从满变为了不满的时候，才需要从条件队列里面唤醒一个线程。

class ConditionalBoundedBufer<V> extends BaseBoundedBuffer<V>{
    protected ConditionalBoundedBufer(int size) {
        super(size);
    }
    public synchronized void put(V v) throws InterruptedException {
        while(isFull()){
            wait();
        }
        boolean isEmpty = isEmpty();
        doPut(v);
        if(isEmpty){//从空变为了非空的时候，才需要唤醒（而实际上需要唤醒那些take线程，而不是put线程）
            notifyAll();
        }
    }
    public synchronized V take() throws InterruptedException {
        while(isEmpty()){
            wait();
        }
        boolean isFull = isFull();
        V v = doTake();
        if(isFull){//从满变为了不满，才需要唤醒（而实际上需要唤醒那些put线程，而不是take线程）
            notifyAll();
        }
        return v;
    }
}

这只是一种小技巧，会加大程序的复杂性，不提倡！
从空变为了非空，唤醒的应该是那些阻塞在take()上的，从满变为了不满唤醒的应该是那些阻塞在put()上的线程，而notifyAll会把所有条件队列里面的所有的等待的线程全部唤醒，这就显现出了内部条件队列有一个缺陷：内部锁只能有一个与之关联的条件队列。显式的condition的出现就是为了解决这个问题。
正如Lock提供了比内部锁更丰富的特征一样，condition也提供了比内部条件队列更丰富更灵活的功能。一个lock可以有多个condition，一个condition只关联到一个Lock。

class ConditionBoundedBufer<T> {//使用显式的条件变量，HLL的登场了
    private Lock lock = new ReentrantLock();
    private Condition notEmpty = lock.newCondition();
    private Condition notFull = lock.newCondition();
    private final T[] items = (T[])new Object[100];
    private int head,tail,count;
    
    //阻塞，一直到notFull
    public void put(T t) throws InterruptedException {
        lock.lock();
        try{
            while(count == items.length){
                notFull.await();//等待非满
            }
            items[tail] = t;
            tail ++;
            if(tail == items.length){
                tail = 0;
            }
            count++;
            notEmpty.signal();//唤醒那些执行take()而阻塞的线程
            
        }finally{
            lock.unlock();
        }
    }
    //阻塞，一直到notEmpty
    public T take() throws InterruptedException {
        lock.lock();
        try{
            while(count == 0){
                notEmpty.await();//等待非空
            }
            T t = items[head];
            items[head] = null;
            head ++;
            if(head == items.length){
                head = 0;
            }
            count--;
            notFull.signal();//唤醒那些执行put()而阻塞的线程
            return t;
        }finally{
            lock.unlock();
        }
    }
}

至此，上面的所有的问题已经全部完美的得到了解决！

希望以上对你理解wait&notify,lock&condition有所帮助，也欢迎大家观看我的两个视频课程：Java生产环境下性能监控与调优详解 Java秒杀系统方案优化 高性能高并发实战