文档章节

BlockingQueue深入分析

vshcxl
 vshcxl
发布于 2016/12/12 22:13
字数 1315
阅读 14
收藏 1

1.BlockingQueue定义的常用方法如下

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

 

1)add(anObject):把anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则招聘异常

2)offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.

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

4)poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null

5)take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止

其中:BlockingQueue 不接受null 元素。试图add、put 或offer 一个null 元素时,某些实现会抛出NullPointerException。null 被用作指示poll 操作失败的警戒值。 

 

2、BlockingQueue的几个注意点

【1】BlockingQueue 可以是限定容量的。它在任意给定时间都可以有一个remainingCapacity,超出此容量,便无法无阻塞地put 附加元素。没有任何内部容量约束的BlockingQueue 总是报告Integer.MAX_VALUE 的剩余容量。

【2】BlockingQueue 实现主要用于生产者-使用者队列,但它另外还支持Collection 接口。因此,举例来说,使用remove(x) 从队列中移除任意一个元素是有可能的。然而,这种操作通常 会有效执行,只能有计划地偶尔使用,比如在取消排队信息时。

【3】BlockingQueue 实现是线程安全的。所有排队方法都可以使用内部锁或其他形式的并发控制来自动达到它们的目的。然而,大量的 Collection 操作(addAll、containsAll、retainAll 和removeAll)没有 必要自动执行,除非在实现中特别说明。因此,举例来说,在只添加了c 中的一些元素后,addAll(c) 有可能失败(抛出一个异常)。

【4】BlockingQueue 实质上不支持使用任何一种“close”或“shutdown”操作来指示不再添加任何项。这种功能的需求和使用有依赖于实现的倾向。例如,一种常用的策略是:对于生产者,插入特殊的end-of-stream 或poison 对象,并根据使用者获取这些对象的时间来对它们进行解释。

 

3、简要概述BlockingQueue常用的四个实现类

 

 

1)ArrayBlockingQueue:规定大小的BlockingQueue,其构造函数必须带一个int参数来指明其大小.其所含的对象是以FIFO(先入先出)顺序排序的.

2)LinkedBlockingQueue:大小不定的BlockingQueue,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定.其所含的对象是以FIFO(先入先出)顺序排序的

3)PriorityBlockingQueue:类似于LinkedBlockQueue,但其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序.

4)SynchronousQueue:特殊的BlockingQueue,对其的操作必须是放和取交替完成的.

    

其中LinkedBlockingQueue和ArrayBlockingQueue比较起来,它们背后所用的数据结构不一样,导致LinkedBlockingQueue的数据吞吐量要大于ArrayBlockingQueue,但在线程数量很大时其性能的可预见性低于ArrayBlockingQueue.  

 

下面主要看一下ArrayBlockingQueue的源码:


 

public boolean offer(E e) {      
        if (e == null) throw new NullPointerException();      
        final ReentrantLock lock = this.lock;//每个对象对应一个显示的锁      
        lock.lock();//请求锁直到获得锁(不可以被interrupte)      
        try {      
            if (count == items.length)//如果队列已经满了      
                return false;      
            else {      
                insert(e);      
                return true;      
            }      
        } finally {      
            lock.unlock();//      
        }      
}      
看insert方法:      
private void insert(E x) {      
        items[putIndex] = x;      
        //增加全局index的值。      
        /*    
        Inc方法体内部:    
        final int inc(int i) {    
        return (++i == items.length)? 0 : i;    
            }    
        这里可以看出ArrayBlockingQueue采用从前到后向内部数组插入的方式插入新元素的。如果插完了,putIndex可能重新变为0(在已经执行了移除操作的前提下,否则在之前的判断中队列为满)    
        */     
        putIndex = inc(putIndex);       
        ++count;      
        notEmpty.signal();//wake up one waiting thread      
}  
public void put(E e) throws InterruptedException {      
        if (e == null) throw new NullPointerException();      
        final E[] items = this.items;      
        final ReentrantLock lock = this.lock;      
        lock.lockInterruptibly();//请求锁直到得到锁或者变为interrupted      
        try {      
            try {      
                while (count == items.length)//如果满了,当前线程进入noFull对应的等waiting状态      
                    notFull.await();      
            } catch (InterruptedException ie) {      
                notFull.signal(); // propagate to non-interrupted thread      
                throw ie;      
            }      
            insert(e);      
        } finally {      
            lock.unlock();      
        }      
}  
public boolean offer(E e, long timeout, TimeUnit unit)      
        throws InterruptedException {      
     
        if (e == null) throw new NullPointerException();      
    long nanos = unit.toNanos(timeout);      
        final ReentrantLock lock = this.lock;      
        lock.lockInterruptibly();      
        try {      
            for (;;) {      
                if (count != items.length) {      
                    insert(e);      
                    return true;      
                }      
                if (nanos <= 0)      
                    return false;      
                try {      
                //如果没有被 signal/interruptes,需要等待nanos时间才返回      
                    nanos = notFull.awaitNanos(nanos);      
                } catch (InterruptedException ie) {      
                    notFull.signal(); // propagate to non-interrupted thread      
                    throw ie;      
                }      
            }      
        } finally {      
            lock.unlock();      
        }      
    }      
public boolean add(E e) {      
    return super.add(e);      
}      
父类:      
public boolean add(E e) {      
        if (offer(e))      
            return true;      
        else     
            throw new IllegalStateException("Queue full");      
    }    

该类中有几个实例变量:takeIndex/putIndex/count

用三个数字来维护这个队列中的数据变更:      
    /** items index for next take, poll or remove */     
    private int takeIndex;      
    /** items index for next put, offer, or add. */     
    private int putIndex;      
    /** Number of items in the queue */     
    private int count;    

 

本文转载自:http://blog.csdn.net/tayanxunhua/article/details/20962307

共有 人打赏支持
vshcxl
粉丝 22
博文 283
码字总数 34755
作品 0
浦东
高级程序员
私信 提问
集合框架 Queue---BlockingQueue详解

本例介绍一个特殊的队列:BlockingQueue,如果BlockingQueue是空的,从BlockingQueue取东西的操作将会被阻断进入等待状态,直到BlockingQueue进了东西才会被唤醒,同样,如果BlockingQueue是...

长平狐
2012/11/28
587
0
BlockingQueue详解

本例再次实现前面介绍的篮子程序,不过这个篮子中最多能放得苹果数不是1,可以随意指定。当篮子满时,生产者进入等待状态,当篮子空时,消费者等待。 BlockingQueue定义的常用方法如下: ad...

wow863597
2016/05/09
53
0
JDK5.0新特性系列---11.6线程 BlockingQueue

importjava.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /**......

tavenli
2011/12/17
0
0
java 队列阻塞方法ArrayBlockingQueue学习

本例介绍一个特殊的队列:BlockingQueue,如果BlockQueue是空的,从BlockingQueue取东西的操作将会被阻断进入等待状态,直到BlockingQueue进了东西才会被唤醒.同样,如果BlockingQueue是满的,任何...

孟飞阳
2016/07/01
23
0
使用BlockingQueue的生产者消费者模式

BlockingQueue很好的解决了多线程中,如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利。使用场景。 首先它是一个队列,...

雪童子
2016/04/24
0
0

没有更多内容

加载失败,请刷新页面

加载更多

码云项目100,水一发

简单回顾一下: 早期构想最多的,是希望能将PHP一些类和编码分区做得更细,所以很多尝试。但不得不说,PHP的功能过于单一,是的,也许写C/C++扩展,可以解决问题,那我为什么不用C#或者Golan...

曾建凯
今天
3
0
Spring应用学习——AOP

1. AOP 1. AOP:即面向切面编程,采用横向抽取机制,取代了传统的继承体系的重复代码问题,如下图所示,性能监控、日志记录等代码围绕业务逻辑代码,而这部分代码是一个高度重复的代码,也就...

江左煤郎
今天
4
0
eclipse的版本

Eclipse各版本代号一览表 Eclipse的设计思想是:一切皆插件。Eclipse核心很小,其它所有功能都以插件的形式附加于Eclipse核心之上。 Eclipse基本内核包括:图形API(SWT/Jface),Java开发环...

mdoo
今天
3
0
SpringBoot源码:启动过程分析(一)

本文主要分析 SpringBoot 的启动过程。 SpringBoot的版本为:2.1.0 release,最新版本。 一.时序图 还是老套路,先把分析过程的时序图摆出来:时序图-SpringBoot2.10启动分析 二.源码分析 首...

Jacktanger
今天
6
0
小白带你认识netty(二)之netty服务端启动(上)

上一章 中的标准netty启动代码中,ServerBootstrap到底是如何启动的呢?这一章我们来瞅下。 server.group(bossGroup, workGroup);server.channel(NioServerSocketChannel.class).optio...

天空小小
今天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部