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 中更靈活、更具可伸縮性的鎖定機制