[置顶] 生产者消费者模型(二)-引入ArrayBlockingQueue

时间：2021-08-11 17:40:13

前言        

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


Queue

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

[置顶] 生产者消费者模型(二)-引入ArrayBlockingQueue


代码展示

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

Producer.java

/**
 * 生产者
 * @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

 
/**
 * 消费者
 * @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

 

/**
 * 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          在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这些容器可能知道的就是太多了。我在这里也提供一个入口。至于这个模型我觉得更需要关心的是其中的思想，而不是实现本身。

标签：

相关文章