生产者消费者模型(二)-引入ArrayBlockingQueue

前言        

        在《生产者消费者模型你知道多少》中简单的模拟了一个生产者消费者模型。有些网友对我的实现提出了很多质疑。我在文章的结尾也对抛出了一个问题：在添加的过程中可能出现数据丢失的情况，应该怎么处理？在代码中也充斥了大量的锁，有些锁是不需要的，而且有重复制造轮子的嫌疑。在今天我将引入ArrayBlockingQueue重写这个模型，这在实际开发中可能更有意义，另外对于java.util.concurrent这个包有一个认识。这个包在并发编程实践里面有非常重要的作用。模型我只会把代码贴出来
，不再做具体的分析，因为在第一篇中有关于思想性的东西讲的比较多。我会重点讲一下ArrayBlockingQueue，举ArrayBlockingQueue其中的几个方法进行简要的分析。

Queue

        在写代码之前我先给出java.util.concurrent包中所有Queue的UML结构图，有兴趣的同学也可以去看其中的源码，了解其中的一些实现，在实际编程中可以避免重复制造轮子，另一方面也可以学习大牛的设计和代码实现。下面我在写模型中主要会用到其中的ArrayBlockingQueue作为容器（对应的是我上一篇中Container这个类）。这个队列里面就有我今天要写模型里面所有需要的方法。

代码展示

        从代码来看会发现代码量比在第一篇中简少了很多，大量的实现ArrayBlockingQueue已经做掉了，包括判空，线程挂起等操作都封装在ArrayBlockingQueue中。这样具体的应用者的代码会少很多。生产者只需要关心生产，消费者只需要关心消费。而在我的第一篇中具体的生产者还需要去通知消费者，还需要关心整个容器是否满了。从这里可以看出ArrayBlockingQueue是一种比较好的实现方式，高度的内聚，和我的实现有着天壤之别。

Producer.java

[java] view
plain copy

/** 

 * 生产者 

 * @author 百恼 2013-11-16下午07:44:36 

 * 

 */  

public class Producer implements Runnable{  

      

    //容器  

    private final ArrayBlockingQueue<Bread> queue;  

      

    public Producer(ArrayBlockingQueue<Bread> queue){  

        this.queue = queue;  

    }  

  

    /* (non-Javadoc) 

     * @see java.lang.Runnable#run() 

     */  

    @Override  

    public void run() {  

        while(true){  

            produce();  

        }  

    }  

      

    public void produce(){  

        /** 

         * put()方法是如果容器满了的话就会把当前线程挂起 

         * offer()方法是容器如果满的话就会返回false，也正是我在前一篇中实现的那种策略。 

         */  

        try {  

            Bread bread = new Bread();  

            queue.put(bread);  

            System.out.println("Producer:"+bread);  

        } catch (InterruptedException e) {  

            e.printStackTrace();  

        }  

    }  

}  

Consumer.java

[java] view
plain copy

/** 

 * 消费者 

 * @author 百恼  2013-11-16下午07:42:08 

 * 

 */  

public class Consumer implements Runnable{  

      

    //容器  

    private final ArrayBlockingQueue<Bread> queue;  

      

    public Consumer(ArrayBlockingQueue<Bread> queue){  

        this.queue = queue;  

    }  

  

    /* (non-Javadoc) 

     * @see java.lang.Runnable#run() 

     */  

    @Override  

    public void run() {  

        while(true){  

            consume();  

        }  

    }  

      

    public void consume(){  

        /** 

         * take()方法和put()方法是对应的，从中拿一个数据，如果拿不到线程挂起 

         * poll()方法和offer()方法是对应的，从中拿一个数据，如果没有直接返回null 

         */  

        try {  

            Bread bread = queue.take();  

            System.out.println("consumer:"+bread);  

        } catch (InterruptedException e) {  

            e.printStackTrace();  

        }  

    }  

}  

Client.java

[java] view
plain copy

/** 

 * TODO Comment of Client 

 * @author 百恼 2013-11-16下午07:58:38 

 * 

 */  

public class Client {  

  

    /** 

     * @param args 

     */  

    public static void main(String[] args) {  

        int capacity = 10;  

        ArrayBlockingQueue<Bread> queue = new ArrayBlockingQueue<Bread>(capacity);  

  

        new Thread(new Producer(queue)).start();  

        new Thread(new Producer(queue)).start();  

        new Thread(new Consumer(queue)).start();  

        new Thread(new Consumer(queue)).start();  

        new Thread(new Consumer(queue)).start();  

    }  

  

}  

ArrayBlockingQueue简要分析

ConditionObject

        如果看了源码的可以看出ArrayBlockingQueue并没有用List作为容器，而是用了数组。我在第一篇中的实现就直接用了List去实现。这样肯定是为了性能去考虑。另外一个就是线程监听器的实现用的Condition，具体的是ConditionObject这个类。核心是LockSupport.park()和LockSupport.unpark()(参考LockSupport源码分析)。

remoteAt方法

         


         在ArrayBlockingQueue里面用了一个for循环然后删除节点为i的数据，如果还记得ArrayList的remove会发现，它里面直接用了System.arraycopy()方法进行删除其中的一条数据。谁的性能好一些，也没有实测过，在这里就不妄下结论。

poll方法

        外在ArrayBlockingQueue中有这个poll(long timeout, TimeUnit unit)这个方法，这个方法是传入一个参数就是等待timeout ms后返回。实际应用中会发现这个等待时间是不准确的。我在一次对接口测试的过程中发现了这样一个问题。

总结

       这一篇中主要讲了用ArrayBlockingQueue实现生产者消费者模型，这个实现的本身意义不大，更大的意义是带入到java.util.concurrent包中去。可能对于一些人来说这个包比较陌生，即使用过这个包去编程的人也可能只限于AtomicInteger,ArrayBlockingQueue,CountDownLatch这些。就好像容器常用的也就ArrayList,HashSet,HashMap这些一样。对于TreeMap，ConcurrentHashMap这些容器可能知道的就是太多了。我在这里也提供一个入口。至于这个模型我觉得更需要关心的是其中的思想，而不是实现本身。