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

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

       public class BlockOperator {private Object notEmpty = new Object();private Queue&lt;Object&gt; linkedList = new LinkedList&lt;Object&gt;();/** * 取物品 *  * @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;}@Overridepublic void run() {//生产n个产品for(int i=0;i&lt;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;}@Overridepublic 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产生的。快下班了，写的快，刚才那个问题没太细想，高手们可以帮助出解决方案。
2 楼 liaohexiang 2011-08-27 锁的粒度太大了，producer 和 consumer 在同一时间只能有一个进入queue中，可以看看concurrent包里的BlockingQueue 3 楼 wmj007 2011-09-13 学习思路自己扩展

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

热点推荐