码迷，mamicode.com
首页 > 编程语言 > 详细

java 多线程阻塞队列 与 阻塞方法与和非阻塞方法

时间：2017-03-23 19:07:25      阅读：240      评论：0      收藏：0      [点我收藏+]

标签：zed   throws   index   多线程   调用   jdk   capacity   ide   ring   

Queue是什么

队列，是一种数据结构。除了优先级队列和LIFO队列外，队列都是以FIFO（先进先出）的方式对各个元素进行排序的。无论使用哪种排序方式，队列的头都是调用remove()或poll()移除元素的。在FIFO队列中，所有新元素都插入队列的末尾。队列都是线程安全的，内部已经实现安全措施，不用我们担心

 

Queue中的方法

Queue中的方法不难理解，6个，每2对是一个也就是总共3对。看一下JDK API就知道了：

技术分享

注意一点就好，Queue通常不允许插入Null，尽管某些实现（比如LinkedList）是允许的，但是也不建议。



使用非阻塞队列的时候有一个很大问题就是：它不会对当前线程产生阻塞，那么在面对类似消费者-生产者的模型时，就必须额外地实现同步策略以及线程间唤醒策略，这个实现起来就非常麻烦。但是有了阻塞队列就不一样了，它会对当前线程产生阻塞，比如一个线程从一个空的阻塞队列中取元素，此时线程会被阻塞直到阻塞队列中有了元素。当队列中有元素后，被阻塞的线程会自动被唤醒（不需要我们编写代码去唤醒）。这样提供了极大的方便性。

一.几种主要的阻塞队列

　自从Java 1.5之后，在java.util.concurrent包下提供了若干个阻塞队列，主要有以下几个：

　　ArrayBlockingQueue：基于数组实现的一个阻塞队列，在创建ArrayBlockingQueue对象时必须制定容量大小。并且可以指定公平性与非公平性，默认情况下为非公平的，即不保证等待时间最长的队列最优先能够访问队列。

　　LinkedBlockingQueue：基于链表实现的一个阻塞队列，在创建LinkedBlockingQueue对象时如果不指定容量大小，则默认大小为Integer.MAX_VALUE。

　　PriorityBlockingQueue：以上2种队列都是先进先出队列，而PriorityBlockingQueue却不是，它会按照元素的优先级对元素进行排序，按照优先级顺序出队，每次出队的元素都是优先级最高的元素。注意，此阻塞队列为无界阻塞队列，即容量没有上限（通过源码就可以知道，它没有容器满的信号标志），前面2种都是有界队列。

　　DelayQueue：基于PriorityQueue，一种延时阻塞队列，DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue也是一个无界队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。


二.阻塞队列中的方法 VS 非阻塞队列中的方法
 
1.非阻塞队列中的几个主要方法：
 
　　add(E e):将元素e插入到队列末尾，如果插入成功，则返回true；如果插入失败（即队列已满），则会抛出异常；
 
　　remove()：移除队首元素，若移除成功，则返回true；如果移除失败（队列为空），则会抛出异常；
 
　　offer(E e)：将元素e插入到队列末尾，如果插入成功，则返回true；如果插入失败（即队列已满），则返回false；
 
　　poll()：移除并获取队首元素，若成功，则返回队首元素；否则返回null；
 
　　peek()：获取队首元素，若成功，则返回队首元素；否则返回null
 

　　对于非阻塞队列，一般情况下建议使用offer、poll和peek三个方法，不建议使用add和remove方法。因为使用offer、poll和peek三个方法可以通过返回值判断操作成功与否，而使用add和remove方法却不能达到这样的效果。注意，非阻塞队列中的方法都没有进行同步措施。

//在这篇笔记中没有介绍非阻塞队列，大部分阻塞队列都可以有非阻塞方法和阻塞方法
 
2.阻塞队列中的几个主要方法：
 
　　阻塞队列包括了非阻塞队列中的大部分方法，上面列举的5个方法在阻塞队列中都存在，但是要注意这5个方法在阻塞队列中都进行了同步措施。除此之外，阻塞队列提供了另外4个非常有用的方法：
 
　　put(E e)
 
　　take()
 
　　offer(E e,long timeout, TimeUnit unit)
 
　　poll(long timeout, TimeUnit unit)
 
 
 
　　put方法用来向队尾存入元素，如果队列满，则等待；
 
　　take方法用来从队首取元素，如果队列为空，则等待；
 
　　offer方法用来向队尾存入元素，如果队列满，则等待一定的时间，当时间期限达到时，如果还没有插入成功，则返回false；否则返回true；
 
　　poll方法用来从队首取元素，如果队列空，则等待一定的时间，当时间期限达到时，如果取到，则返回null；否则返回取得的元素；
 

注意：

1、必须要使用take()方法在获取的时候达成阻塞结果
2、使用poll()方法将产生非阻塞效果


三.阻塞队列的实现原理
 
　　前面谈到了非阻塞队列和阻塞队列中常用的方法，下面来探讨阻塞队列的实现原理，本文以ArrayBlockingQueue为例，其他阻塞队列实现原理可能和ArrayBlockingQueue有一些差别，但是大体思路应该类似，有兴趣的朋友可自行查看其他阻塞队列的实现源码。
 
　　首先看一下ArrayBlockingQueue类中的几个成员变量：

  1. public class ArrayBlockingQueue<E> extends AbstractQueue<E>
  2. implements BlockingQueue<E>, java.io.Serializable {
  3.  
  4. private static final long serialVersionUID = -817911632652898426L;
  5.  
  6. /** The queued items  */
  7. private final E[] items;
  8. /** items index for next take, poll or remove */
  9. private int takeIndex;
  10. /** items index for next put, offer, or add. */
  11. private int putIndex;
  12. /** Number of items in the queue */
  13. private int count;
  14.  
  15. /*
  16. * Concurrency control uses the classic two-condition algorithm
  17. * found in any textbook.
  18. */
  19.  
  20. /** Main lock guarding all access */
  21. private final ReentrantLock lock;
  22. /** Condition for waiting takes */
  23. private final Condition notEmpty;
  24. /** Condition for waiting puts */
  25. private final Condition notFull;
  26. }

可以看出，ArrayBlockingQueue中用来存储元素的实际上是一个数组，takeIndex和putIndex分别表示队首元素和队尾元素的下标，count表示队列中元素的个数。
 
　　lock是一个可重入锁，notEmpty和notFull是等待条件。
 
　　下面看一下ArrayBlockingQueue的构造器，构造器有三个重载版本：
  2. public ArrayBlockingQueue(int capacity) {
  3. }
  4. public ArrayBlockingQueue(int capacity, boolean fair) {
  5.  
  6. }
  7. public ArrayBlockingQueue(int capacity, boolean fair,
  8.                           Collection<? extends E> c) {
  9. }

　　第一个构造器只有一个参数用来指定容量，第二个构造器可以指定容量和公平性，第三个构造器可以指定容量、公平性以及用另外一个集合进行初始化。
 
　　然后看它的两个关键方法的实现：put()和take()：
 
  1. public void put(E e) throws InterruptedException {
  2.     if (e == null) throw new NullPointerException();
  3.     final E[] items = this.items;
  4.     final ReentrantLock lock = this.lock;
  5.     lock.lockInterruptibly();
  6.     try {
  7.         try {
  8.             while (count == items.length)
  9.                 notFull.await();
  10.         } catch (InterruptedException ie) {
  11.             notFull.signal(); // propagate to non-interrupted thread
  12.             throw ie;
  13.         }
  14.         insert(e);
  15.     } finally {
  16.         lock.unlock();
  17.     }
  18. }

　　从put方法的实现可以看出，它先获取了锁，并且获取的是可中断锁，然后判断当前元素个数是否等于数组的长度，如果相等，则调用notFull.await()进行等待，如果捕获到中断异常，则唤醒线程并抛出异常。
 
　　当被其他线程唤醒时，通过insert(e)方法插入元素，最后解锁。
 
　　我们看一下insert方法的实现：
  2. private void insert(E x) {
  3.     items[putIndex] = x;
  4.     putIndex = inc(putIndex);
  5.     ++count;
  6.     notEmpty.signal();
  7. }

　　它是一个private方法，插入成功后，通过notEmpty唤醒正在等待取元素的线程。
 
　　下面是take()方法的实现：
 
  1. public E take() throws InterruptedException {
  2.     final ReentrantLock lock = this.lock;
  3.     lock.lockInterruptibly();
  4.     try {
  5.         try {
  6.             while (count == 0)
  7.                 notEmpty.await();
  8.         } catch (InterruptedException ie) {
  9.             notEmpty.signal(); // propagate to non-interrupted thread
  10.             throw ie;
  11.         }
  12.         E x = extract();
  13.         return x;
  14.     } finally {
  15.         lock.unlock();
  16.     }
  17. }

　　跟put方法实现很类似，只不过put方法等待的是notFull信号，而take方法等待的是notEmpty信号。在take方法中，如果可以取元素，则通过extract方法取得元素，下面是extract方法的实现：
  2. private E extract() {
  3.     final E[] items = this.items;
  4.     E x = items[takeIndex];
  5.     items[takeIndex] = null;
  6.     takeIndex = inc(takeIndex);
  7.     --count;
  8.     notFull.signal();
  9.     return x;
  10. }

　　跟insert方法也很类似。
 
　　其实从这里大家应该明白了阻塞队列的实现原理，事实它和我们用Object.wait()、Object.notify()和非阻塞队列实现生产者-消费者的思路类似，只不过它把这些工作一起集成到了阻塞队列中实现。


四.示例和使用场景
 
　　下面先使用Object.wait()和Object.notify()、非阻塞队列实现生产者-消费者模式：
 
  1. public class Test {
  2.     private int queueSize = 10;
  3.     private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
  4.  
  5.     public static void main(String[] args)  {
  6.         Test test = new Test();
  7.         Producer producer = test.new Producer();
  8.         Consumer consumer = test.new Consumer();
  9.  
  10.         producer.start();
  11.         consumer.start();
  12.     }
  13.  
  14.     class Consumer extends Thread{
  15.  
  16.         @Override
  17.         public void run() {
  18.             consume();
  19.         }
  20.  
  21.         private void consume() {
  22.             while(true){
  23.                 synchronized (queue) {
  24.                     while(queue.size() == 0){
  25.                         try {
  26.                             System.out.println("队列空，等待数据");
  27.                             queue.wait();
  28.                         } catch (InterruptedException e) {
  29.                             e.printStackTrace();
  30.                             queue.notify();
  31.                         }
  32.                     }
  33.                     queue.poll();          //每次移走队首元素
  34.                     queue.notify();
  35.                     System.out.println("从队列取走一个元素，队列剩余"+queue.size()+"个元素");
  36.                 }
  37.             }
  38.         }
  39.     }
  40.  
  41.     class Producer extends Thread{
  42.  
  43.         @Override
  44.         public void run() {
  45.             produce();
  46.         }
  47.  
  48.         private void produce() {
  49.             while(true){
  50.                 synchronized (queue) {
  51.                     while(queue.size() == queueSize){
  52.                         try {
  53.                             System.out.println("队列满，等待有空余空间");
  54.                             queue.wait();
  55.                         } catch (InterruptedException e) {
  56.                             e.printStackTrace();
  57.                             queue.notify();
  58.                         }
  59.                     }
  60.                     queue.offer(1);        //每次插入一个元素
  61.                     queue.notify();
  62.                     System.out.println("向队列取中插入一个元素，队列剩余空间："+(queueSize-queue.size()));
  63.                 }
  64.             }
  65.         }
  66.     }
  67. }

　　这个是经典的生产者-消费者模式，通过阻塞队列和Object.wait()和Object.notify()实现，wait()和notify()主要用来实现线程间通信。
 
　　具体的线程间通信方式（wait和notify的使用）在后续问章中会讲述到。
 
　　下面是使用阻塞队列实现的生产者-消费者模式：

  1. public class Test {
  2.     private int queueSize = 10;
  3.     private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(queueSize);
  4.  
  5.     public static void main(String[] args)  {
  6.         Test test = new Test();
  7.         Producer producer = test.new Producer();
  8.         Consumer consumer = test.new Consumer();
  9.  
  10.         producer.start();
  11.         consumer.start();
  12.     }
  13.  
  14.     class Consumer extends Thread{
  15.  
  16.         @Override
  17.         public void run() {
  18.             consume();
  19.         }
  20.  
  21.         private void consume() {
  22.             while(true){
  23.                 try {
  24.                     queue.take();
  25.                     System.out.println("从队列取走一个元素，队列剩余"+queue.size()+"个元素");
  26.                 } catch (InterruptedException e) {
  27.                     e.printStackTrace();
  28.                 }
  29.             }
  30.         }
  31.     }
  32.  
  33.     class Producer extends Thread{
  34.  
  35.         @Override
  36.         public void run() {
  37.             produce();
  38.         }
  39.  
  40.         private void produce() {
  41.             while(true){
  42.                 try {
  43.                     queue.put(1);
  44.                     System.out.println("向队列取中插入一个元素，队列剩余空间："+(queueSize-queue.size()));
  45.                 } catch (InterruptedException e) {
  46.                     e.printStackTrace();
  47.                 }
  48.             }
  49.         }
  50.     }
  51. }


　有没有发现，使用阻塞队列代码要简单得多，不需要再单独考虑同步和线程间通信的问题。
 
　　在并发编程中，一般推荐使用阻塞队列，这样实现可以尽量地避免程序出现意外的错误。
 
　　阻塞队列使用最经典的场景就是socket客户端数据的读取和解析，读取数据的线程不断将数据放入队列，然后解析线程不断从队列取数据解析。还有其他类似的场景，只要符合生产者-消费者模型的都可以使用阻塞队列。


java 多线程阻塞队列 与 阻塞方法与和非阻塞方法

标签：zed   throws   index   多线程   调用   jdk   capacity   ide   ring   

原文地址：http://www.cnblogs.com/signheart/p/6606475.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！