Java可阻塞队列的两种实现方式

在Java中，对于Lock和Condition能够理解为对传统的synchronized和wait/notify机制的替代。

wait/notify有个限制，调用wait/notify的线程必须持有对象的锁。

This method should only be called by a thread that is the owner of this object's monitor. See the notify method for a description of the ways in which a thread can become the owner of a monitor.

Throws:

IllegalMonitorStateException - if the current thread is not the owner of this object's monitor.

一般使用wait/notify的代码是这个样子的：

synchronized (obj) {
         while (<condition does not hold>)
             obj.wait();
         ... // Perform action appropriate to condition
     }

在Condition接口的javadoc中，有一个经典的Condition例子，用Condition实现了一个可阻塞队列。这里仿照javadoc简单实现了一个可阻塞队列。为了简单，没有进行try/catch，同时加入了一些注释。

<!-- lang: java -->
public class BoundedBuffer {
    final Lock lock = new ReentrantLock();
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();

final Object[] items = new Object[2]; // 阻塞队列
int putptr, takeptr, count;

public void put(Object x) throws InterruptedException {
	System.out.println("进入put");
	lock.lock();
	System.out.println("put lock 锁住");
	try {
		while (count == items.length) { // 若是队列满了，notFull就一直等待
			System.out.println("put notFull 等待");
			notFull.await(); // 调用await的意思取反，及not notFull -> Full
		}
		items[putptr] = x; // 终于能够插入队列
		if (++putptr == items.length)
			putptr = 0; // 若是下标到达数组边界，循环下标置为0
		++count;
		System.out.println("put notEmpty 唤醒");
		notEmpty.signal(); // 唤醒notEmpty
	} finally {
		System.out.println("put lock 解锁");
		lock.unlock();
	}
}

public Object take() throws InterruptedException {
	lock.lock();
	System.out.println("take lock 锁住");
	try {
		while (count == 0) {
			System.out.println("take notEmpty 等待");
			notEmpty.await();
		}
		Object x = items[takeptr];
		if (++takeptr == items.length)
			takeptr = 0;
		--count;
		System.out.println("take notFull 唤醒");
		notFull.signal();
		return x;
	} finally {
		lock.unlock();
		System.out.println("take lock 解锁");
	}
}

public static void main(String[] args) throws InterruptedException {
	final BoundedBuffer bb = new BoundedBuffer();
	System.out.println(Thread.currentThread()+","+bb);
	
	new Thread(new Runnable() {
		@Override
		public void run() {
			try {
				Thread.sleep(1000);
				System.out.println(Thread.currentThread()+","+bb);
				bb.put("xx");
				bb.put("yy");
				bb.put("zz");
                                    bb.put("zz");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}).start();
	bb.take();
}
}

若是不使用JUC，大概是这样的：

<!-- lang: java -->
public class BoundedBuffer_Synchronized {
private Object[] items = new Object[2];
private Object notEmpty = new Object();
private Object notFull = new Object();
int count,putidx,takeidx;

public  void put(Object obj) throws InterruptedException{
	synchronized(notFull){
		while(count == items.length){
			notFull.wait();
		}
	}
	items[putidx] = obj;
	if(++putidx == items.length){
		putidx = 0;
	}
	count ++;
	synchronized (notEmpty) {
		notEmpty.notify();
	}
}
public Object take() throws InterruptedException{
	synchronized(notEmpty){
		while(count == 0){ // 啥也没有呢 取啥
			notEmpty.wait();
		}
	}
	Object x = items[takeidx];
	System.out.println("取第"+takeidx+"个元素"+x);
	if(++takeidx == items.length){
		takeidx = 0; 
	}
	count --;
	synchronized (notFull) {
		notFull.notify();
	}
	return x;
}
public static void main(String[] args) throws InterruptedException {
	final BoundedBuffer_Synchronized bb = new BoundedBuffer_Synchronized();
	System.out.println(Thread.currentThread()+","+bb);
	
	new Thread(new Runnable() {
		@Override
		public void run() {
			try {
				Thread.sleep(1000);
				System.out.println(Thread.currentThread()+","+bb);
				bb.put("xx");
				bb.put("yy");
				bb.put("zz");
				bb.put("zz");
				bb.put("zz");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}).start();
	bb.take();
	bb.take();
}
}

从功能上来说，二者实现了可阻塞队列的基本业务需求。Condition是配合Lock使用的，而wait/notify是配合synchronized使用的。比较两种实现方式，其实就是比较Lock和synchronized两种同步机制的区别。关于这方面，能够参考Java 理论与实践: JDK 5.0 中更灵活、更具可伸缩性的锁定机制