Java BlockingQueue的使用


概念介绍:

        阻塞队列,顾名思义,首先它是一个队列,具备队列的一些特点,比如先进先出(FIFO):先插入队列的元素也最先出队列,类似于排队功能。

        阻塞体现在:如果BlockQueue是空的,从BlockQueue中取东西的操作将会被阻断进入等待状态,直到BlockQueue进了东西才会被唤醒;同样如果BlockingQueue是满的,任何试图往里存东西的操作也会被阻断进入等待状态,直到BlockingQueue里有空间才会被唤醒继续操作. 


BlockingQueue常用方法解析:

Action 抛出异常 特殊值 阻塞 超时
插入 add(e) offer(e) put(e) offer(e,time,unit)
取出 remove() poll() take() poll(time,unit)
检查 element() peek()


插入队列数据方法

        add(e):把一个object加到BlockingQueue里,如果BlockingQueue可以容纳,返回true,失败抛出IllegalStateException("Deque full");其实通过看源码发现,add()方法其实最终调用的是offer(e)方法,自上层判断如果返回值是false,扔出异常。

        offer(e):把一个object加到BlockingQueue里,如果BlockingQueue容量未满,返回true;否则返回false;

        put(e) :把一个object加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.

        offfer(e,time,unit):把一个object添加到BlockingQueue里,如果容量未满,返回true;如果队列已满,等待time unit时间,如果能添加成功,返回true;否则返回false;

拉取队列数据方法:

        remove():移除队首元素,如果BlockQueue有数据,返回移除的数据信息;反之则抛出异常。看源码发现,底层其实调用的是poll()方法,上层判断如果poll返回null,则扔出NoSuchElementException()异常;

        注:remove(e):移除一个特定元素,成功返回true;失败返回false;

        poll() :移除队首元素,如果BlockingQueue有数据,返回移除的数据信息;反之则返回null。

        take():移除队首元素,如果BlockingQueue有数据,返回移除的数据信息;反之调用此方法的线程被阻断直到BlocingQueue里面有数据再继续移除。

        poll(time,unit):移除队首元素。返回数据信息和poll()函数一样,区别在于:如果BlockingQueue为空,则等待time unit时间之后,如果队列有数据,则移除,否则返回null。

检查数据方法:

        element():如果BblockingQueue不为空,返回队首元素,但不删除;如果队列为空,则抛出异常,其底层调用的是peekfirst()方法;

        peek():功能和element()一样,但是如果队列为空时,返回null值。


BlockingQueue有5个具体的实现类,根据不同需求,选择不同的实现类 
        1)ArrayBlockingQueue

          基于数组的阻塞队列实现,在ArrayBlockingQueue内部,维护了一个定长数组,以便缓存队列中的数据对象,这是一个常用的阻塞队列,除了一个定长数组外,ArrayBlockingQueue内部还保存着两个整形变量,分别标识着队列的头部和尾部在数组中的位置。
        ArrayBlockingQueue在生产者放入数据和消费者获取数据,都是共用同一个锁对象,由此也意味着两者无法真正并行运行,这点尤其不同于LinkedBlockingQueue;按照实现原理来分析,ArrayBlockingQueue完全可以采用分离锁,从而实现生产者和消费者操作的完全并行运行。Doug Lea之所以没这样去做,也许是因为ArrayBlockingQueue的数据写入和获取操作已经足够轻巧,以至于引入独立的锁机制,除了给代码带来额外的复杂性外,其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不同之处在于,前者在插入或删除元素时不会产生或销毁任何额外的对象实例,而后者则会生成一个额外的Node对象。这在长时间内需要高效并发地处理大批量数据的系统中,其对于GC的影响还是存在一定的区别。而在创建ArrayBlockingQueue时,我们还可以控制对象的内部锁是否采用公平锁,默认采用非公平锁。
        2)LinkedBlockingQueue

          基于链表的阻塞队列,同ArrayListBlockingQueue类似,其内部也维持着一个数据缓冲队列(该队列由一个链表构成),当生产者往队列中放入一个数据时,队列会从生产者手中获取数据,并缓存在队列内部,而生产者立即返回;只有当队列缓冲区达到最大值缓存容量时(LinkedBlockingQueue可以通过构造函数指定该值),才会阻塞生产者队列,直到消费者从队列中消费掉一份数据,生产者线程会被唤醒,反之对于消费者这端的处理也基于同样的原理。而LinkedBlockingQueue之所以能够高效的处理并发数据,还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步,这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据,以此来提高整个队列的并发性能。
         作为开发者,我们需要注意的是,如果构造一个LinkedBlockingQueue对象,而没有指定其容量大小,LinkedBlockingQueue会默认一个类似无限大小的容量(Integer.MAX_VALUE),这样的话,如果生产者的速度一旦大于消费者的速度,也许还没有等到队列满阻塞产生,系统内存就有可能已被消耗殆尽了。
        3)PriorityBlockingQueue

         基于优先级的阻塞队列(优先级的判断通过构造函数传入的Compator对象来决定),但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者,而只会在没有可消费的数据时,阻塞数据的消费者。因此使用的时候要特别注意,生产者生产数据的速度绝对不能快于消费者消费数据的速度,否则时间一长,会最终耗尽所有的可用堆内存空间。在实现PriorityBlockingQueue时,内部控制线程同步的锁采用的是公平锁。


        4)SynchronousQueue

        一种无缓冲的等待队列,类似于无中介的直接交易,有点像原始社会中的生产者和消费者,生产者拿着产品去集市销售给产品的最终消费者,而消费者必须亲自去集市找到所要商品的直接生产者,如果一方没有找到合适的目标,那么对不起,大家都在集市等待。相对于有缓冲的BlockingQueue来说,少了一个中间经销商的环节(缓冲区),如果有经销商,生产者直接把产品批发给经销商,而无需在意经销商最终会将这些产品卖给那些消费者,由于经销商可以库存一部分商品,因此相对于直接交易模式,总体来说采用中间经销商的模式会吞吐量高一些(可以批量买卖);但另一方面,又因为经销商的引入,使得产品从生产者到消费者中间增加了额外的交易环节,单个产品的及时响应性能可能会降低。
        声明一个SynchronousQueue有两种不同的方式,它们之间有着不太一样的行为。公平模式和非公平模式的区别:
        如果采用公平模式:SynchronousQueue会采用公平锁,并配合一个FIFO队列来阻塞多余的生产者和消费者,从而体系整体的公平策略;
        但如果是非公平模式(SynchronousQueue默认):SynchronousQueue采用非公平锁,同时配合一个LIFO队列来管理多余的生产者和消费者,而后一种模式,如果生产者和消费者的处理速度有差距,则很容易出现饥渴的情况,即可能有某些生产者或者是消费者的数据永远都得不到处理。

        

        5) DelayQueue

            DelayQueue中的元素只有当其指定的延迟时间到了,才能够从队列中获取到该元素。DelayQueue是一个没有大小限制的队列,因此往队列中插入数据的操作(生产者)永远不会被阻塞,而只有获取数据的操作(消费者)才会被阻塞。
            使用场景:
            DelayQueue使用场景较少,但都相当巧妙,常见的例子比如使用一个DelayQueue来管理一个超时未响应的连接队列。
注:ArrayBlockingQueue和LinkedBlockingQueue是两个最普通也是最常用的阻塞队列,一般情况下,在处理多线程间的生产者消费者问题,使用这两个类足以。


下面用LinkedBlockQueue实现生产者-消费者模型,代码示例:

package com.bj58.supin.task.sp_statistics; 
/**
 * @author zhaoxin06
 * @Date 2017年3月7日 下午5:17:47
 * @Description 
 *
 */

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;



public class TestBlockQueue{
    
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
        Producer producer1 = new Producer(queue);
        Producer producer2 = new Producer(queue);
        Producer producer3 = new Producer(queue);
        Consumer consumer1 = new Consumer(queue);
        
        ExecutorService service = Executors.newFixedThreadPool(5);
        
        service.execute(producer1);
        service.execute(producer2);
        service.execute(producer3);
        service.execute(consumer1);
        
        //生产者生产10S
        Thread.sleep(10000);
        producer1.stop();
        producer2.stop();
        producer3.stop();
        
        //额外让消费者消费5S
        Thread.sleep(5000);
        //退出Exectors
        service.shutdown();
}
    
    //生产者
    static class Producer implements Runnable{
        BlockingQueue<String> queue = null;
        private volatile boolean isRuning = true;
        private static AtomicInteger count= new AtomicInteger();
        public Producer(BlockingQueue<String> queue){
            this.queue = queue;
        }

        /* (non-Javadoc)
         * @see java.lang.Runnable#run()
         */
        @Override
        public void run() {
            // TODO Auto-generated method stub
            String data = "";
            Random rand = new Random();
            
            System.out.println("生产者线程已启动!");
            try {
                while(isRuning){
                    
                    System.out.println("生产者正在生产数据~");
                    //随机睡眠n毫秒
                    Thread.sleep(rand.nextInt(1000));
                    
                    data ="data:" + count.incrementAndGet();
                    System.out.println("将数据:"+data+"放入队列~");
                    if(!queue.offer(data, 2, TimeUnit.SECONDS)){
                        System.out.println("放入数据失败:"+data);
                    }
                    
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }finally {
                System.out.println("退出生产者线程!");
            }
        }
        
        public void stop(){
            isRuning = false;
        }
        
    }
    
    //消费者
    static class Consumer implements Runnable{
        
        BlockingQueue<String> queue = null;
        public Consumer(BlockingQueue<String> queue){
            this.queue = queue;
        }
        /* (non-Javadoc)
         * @see java.lang.Runnable#run()
         */
        @Override
        public void run() {
            // TODO Auto-generated method stub
            boolean isRuning = true;
            System.out.println("消费者线程启动~~");
            Random r = new Random();
            try {
                while(isRuning){
                    System.out.println("消费者开始消费~");
                    String data = queue.poll(2, TimeUnit.SECONDS);
                    if(data == null){
                        System.out.println("超过2秒钟未获取导数据");
                        isRuning = false;
                    }else{
                        System.out.println("拿到数据:"+data);
                        System.out.println("正在消费数据:"+data);
                        Thread.sleep(r.nextInt(1000));
                    }
                }
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }finally {
                System.out.println("退出消费者线程!");
            }
        }
    }
}


以上是学些BlockingQueue时的一些总结,记录下来,供有需要的人一起学习探讨~~~


暧昧贴