java并发线程-----阻塞队列的实现和思考

     一般来说（单核），如果没有任务会阻塞，那么单处理器机器上使用并发就没有任何意义。   
     阻塞队列多半是解决对同一个（共享）资源进行操作的时候互相协作的问题。比如说有这样一个场景：某个工厂有生产工和搬运工两种角色，前者负责把生产好的产品放入仓库，
后者把仓库里面的产品搬出去卖。假如说仓库的产品数量为零，那么搬运工大可以睡上一觉，等生产工生产出产品，然后再搬。我们抽象出一个实现阻塞队列需要的元素，首先，仓库里面的产品是共享的资源，其次，生产工和搬运工需要互相协调这个资源的产生和消耗，他们应该是对应于程序中的两个线程，互不干扰又互相协调。
     基于这个分析，我们就开始吧。
     首先我们新建阻塞操作类BlockOperator，代码如下：
     

       public class BlockOperator {

	private Object notEmpty = new Object();
	private Queue<Object> linkedList = new LinkedList<Object>();


	/**
	 * 取物品
	 * 
	 * @return
	 * @throws InterruptedException
	 */
 public Object take() throws InterruptedException {
	 synchronized (notEmpty) {
	 String cureadname = Thread.currentThread().getName();
	 System.out.println("搬运工" + cureadname + "来到仓库");
	 sleep(1000l);
	 if (linkedList.size() == 0) {
	 // 假如仓库没东西了，那么就先不取物品，此时释放锁，被唤醒之前，需要先得到锁
	System.out.println("搬运工" + cureadname + "发觉没有物品，只能等待生产");
				notEmpty.wait();
	}
	Object obj = linkedList.poll();//返回并删除此元素
	System.out.println("搬运工" + cureadname + "这时看到有了物品，搬出了:" + obj
					+ "仓库还有物品数量:" + linkedList.size());
		return obj;

}
}

// 生产物品
public void offer(Object object) throws InterruptedException {
     synchronized (notEmpty) {
	String cureadname = Thread.currentThread().getName();
	System.out.println("生产工" + cureadname + "来到仓库准备放物品");
	sleep(3000l);
	if (linkedList.size() == 0) {
	// 假如仓库没东西了，唤醒对象锁。分析：这个时候有可能没有等待锁，也可能有。
	System.out.println("生产工" + cureadname+ "发现来到仓库的时候一件物品都没有，发觉搬运工在睡觉等他或者感觉搬运工在等他，于是喊醒了它");
	notEmpty.notifyAll(); 
	/*
	 * 注 假如仓库有东西，那么不用唤醒搬运工，因为有物品的时候，搬运工不会等待。
	 * 分析：有的人肯定会觉得，有没有这种可能：当linkedList.size=0的时候，notEmpty就wait了，然后在本同步块中，
	 * 发现linkedList.size！=0，那么notEmpty就不会去唤醒了。其实这完全没有可能，因为size！=0只有在完成了
	 * linkedList.add之后才有可能，而在add之前，必然会判断size=0的情况
           */
	}
	System.out.println("生产工" + cureadname + "把物品" + object
					+ "放到了仓库");
	linkedList.add(object);

	}
}

 private void sleep(Long time) {
	try {
		Thread.sleep(time);// 模拟时间消耗
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

     

   
注：本类中，首先创建了一个notEmpty做锁的对象，取物品的时候当产品为0时，notEmpty等待，此时会让位与另一个线程（生产）得到此对象锁，并且唤醒notEmpty上所有的等待锁。在生产完成之后，由于刚才等待的线程已经被唤醒，那么这个搬运工会移除此产品。下面会有我本机运行的效果，对照看一下会一目了然

然后创建生产工线程类CreateProduce，代码如下：

/**
 * 模拟工厂生产产品
 * 
 * @author Administrator
 * 
 */
public class CreateProduce implements Runnable {

	private BlockOperator bq;
	public CreateProduce(BlockOperator bq){
		this.bq=bq;
	}
	
	@Override
	public void run() {
		//生产n个产品
		for(int i=0;i<5;i++){
			Object obj="A"+i;
			try {
				bq.offer(obj);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}		
	}

}

此类比较简单，创建5个商品。

继续创建搬运工线程类，代码如下

/**
 * 模拟工厂搬运产品
 * @author Administrator
 *
 */
public class GetProduce implements Runnable {

	private BlockOperator bq;
	public GetProduce(BlockOperator bq){
		this.bq=bq;
	}
	@Override
	public void run() {
		 try {
			 //搬运工是有多少搬多少
			 while(true){
				 Object obj=bq.take();
			 }
			
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

最后新建一个测试类 如下：

public class TestMain {

	public static void main(String[] args) {
		BlockOperator bq = new BlockOperator();
		Thread createThread = new Thread(new CreateProduce(bq),"creater");
		Thread getThread = new Thread(new GetProduce(bq),"getter");
		createThread.start();
		getThread.start();
		
		
	}

}

在我本地运行的结果如下：


    当我准备想在BlockOperator里面加一个判断产品最大数量的锁对象，比如TheMaxObject，后来在take和offer里面分别加上限制，但是免不了会嵌套synchronized，那么这样会产生死锁。死锁大多是嵌套synchronized产生的。

评论

3 楼 wmj007 2011-09-13  

学习思路自己扩展

2 楼 liaohexiang 2011-08-27  

锁的粒度太大了，producer 和 consumer 在同一时间只能有一个进入queue中，可以看看concurrent包里的BlockingQueue

1 楼 panjun 2011-08-25