您的位置: 首页  >  文章  >  生产者消费者模型(二)-引入ArrayBlockingQueue

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

分类: 文章 • 2024-11-23 22:25:58

前言        

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


Queue

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

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


代码展示

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

Producer.java

[java] view plain copy
  1. /** 
  2.  * 生产者 
  3.  * @author 百恼 2013-11-16下午07:44:36 
  4.  * 
  5.  */  
  6. public class Producer implements Runnable{  
  7.       
  8.     //容器  
  9.     private final ArrayBlockingQueue<Bread> queue;  
  10.       
  11.     public Producer(ArrayBlockingQueue<Bread> queue){  
  12.         this.queue = queue;  
  13.     }  
  14.   
  15.     /* (non-Javadoc) 
  16.      * @see java.lang.Runnable#run() 
  17.      */  
  18.     @Override  
  19.     public void run() {  
  20.         while(true){  
  21.             produce();  
  22.         }  
  23.     }  
  24.       
  25.     public void produce(){  
  26.         /** 
  27.          * put()方法是如果容器满了的话就会把当前线程挂起 
  28.          * offer()方法是容器如果满的话就会返回false，也正是我在前一篇中实现的那种策略。 
  29.          */  
  30.         try {  
  31.             Bread bread = new Bread();  
  32.             queue.put(bread);  
  33.             System.out.println("Producer:"+bread);  
  34.         } catch (InterruptedException e) {  
  35.             e.printStackTrace();  
  36.         }  
  37.     }  
  38. }  

Consumer.java

[java] view plain copy
  1. /** 
  2.  * 消费者 
  3.  * @author 百恼  2013-11-16下午07:42:08 
  4.  * 
  5.  */  
  6. public class Consumer implements Runnable{  
  7.       
  8.     //容器  
  9.     private final ArrayBlockingQueue<Bread> queue;  
  10.       
  11.     public Consumer(ArrayBlockingQueue<Bread> queue){  
  12.         this.queue = queue;  
  13.     }  
  14.   
  15.     /* (non-Javadoc) 
  16.      * @see java.lang.Runnable#run() 
  17.      */  
  18.     @Override  
  19.     public void run() {  
  20.         while(true){  
  21.             consume();  
  22.         }  
  23.     }  
  24.       
  25.     public void consume(){  
  26.         /** 
  27.          * take()方法和put()方法是对应的，从中拿一个数据，如果拿不到线程挂起 
  28.          * poll()方法和offer()方法是对应的，从中拿一个数据，如果没有直接返回null 
  29.          */  
  30.         try {  
  31.             Bread bread = queue.take();  
  32.             System.out.println("consumer:"+bread);  
  33.         } catch (InterruptedException e) {  
  34.             e.printStackTrace();  
  35.         }  
  36.     }  
  37. }  

Client.java

[java] view plain copy
  1. /** 
  2.  * TODO Comment of Client 
  3.  * @author 百恼 2013-11-16下午07:58:38 
  4.  * 
  5.  */  
  6. public class Client {  
  7.   
  8.     /** 
  9.      * @param args 
  10.      */  
  11.     public static void main(String[] args) {  
  12.         int capacity = 10;  
  13.         ArrayBlockingQueue<Bread> queue = new ArrayBlockingQueue<Bread>(capacity);  
  14.   
  15.         new Thread(new Producer(queue)).start();  
  16.         new Thread(new Producer(queue)).start();  
  17.         new Thread(new Consumer(queue)).start();  
  18.         new Thread(new Consumer(queue)).start();  
  19.         new Thread(new Consumer(queue)).start();  
  20.     }  
  21.   
  22. }  

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

相关推荐