一、概述:
二、BlockingQueue定义的常用方法
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 操作失败的警戒值。
三、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 对象,并根据使用者获取这些对象的时间来对它们进行解释。
四、简要概述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.
五、具体BlockingQueue的实现类的内部细节
有耐心的同学请看具体实现类细节:
1、ArrayBlockingQueue
ArrayBlockingQueue是一个由数组支持的有界阻塞队列。此队列按 FIFO(先进先出)原则对元素进行排序。队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素。新元素插入到队列的尾部,队列检索操作则是从队列头部开始获得元素。
这是一个典型的“有界缓存区”,固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区,就不能再增加其容量。试图向已满队列中放入元素会导致放入操作受阻塞;试图从空队列中检索元素将导致类似阻塞。
ArrayBlockingQueue创建的时候需要指定容量capacity(可以存储的最大的元素个数,因为它不会自动扩容)以及是否为公平锁(fair参数)。
在创建ArrayBlockingQueue的时候默认创建的是非公平锁,不过我们可以在它的构造函数里指定。这里调用ReentrantLock的构造函数创建锁的时候,调用了:
public ReentrantLock(boolean fair) {
sync = (fair)? new FairSync() : new NonfairSync();
}
FairSync/ NonfairSync是ReentrantLock的内部类:
线程按顺序请求获得公平锁,而一个非公平锁可以闯入,且当它尚未进入等待队列,就会和等待队列head结点的线程发生竞争,如果锁的状态可用,请求非公平锁的线程可在等待队列中向前跳跃,获得该锁。内部锁synchronized没有提供确定的公平性保证。
分三点来讲这个类:
2.1 添加新元素的方法:add/put/offer
2.2 该类的几个实例变量:takeIndex/putIndex/count/
2.3 Condition实现
1.1 添加新元素的方法:add/put/offer
首先,谈到添加元素的方法,首先得分析以下该类同步机制中用到的锁:
Java代码
- lock = new ReentrantLock(fair);
- notEmpty = lock.newCondition();//Condition Variable 1
- notFull = lock.newCondition();//Condition Variable 2
这三个都是该类的实例变量,只有一个锁lock,然后lock实例化出两个Condition,notEmpty/noFull分别用来协调多线程的读写操作。
Java代码
- 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
- }
Java代码
- 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();
- }
- }
Java代码
- 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();
- }
- }
Java代码
- 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");
- }
1.2 该类的几个实例变量:takeIndex/putIndex/count
Java代码
- 用三个数字来维护这个队列中的数据变更:
- /** 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;
提取元素的三个方法take/poll/remove内部都调用了这个方法:
Java代码
- private E extract() {
- final E[] items = this.items;
- E x = items[takeIndex];
- items[takeIndex] = null;//移除已经被提取出的元素
- takeIndex = inc(takeIndex);//策略和添加元素时相同
- --count;
- notFull.signal();//提醒其他在notFull这个Condition上waiting的线程可以尝试工作了
- return x;
- }
从这个方法里可见,tabkeIndex维护一个可以提取/移除元素的索引位置,因为takeIndex是从0递增的,所以这个类是FIFO队列。
putIndex维护一个可以插入的元素的位置索引。
count显然是维护队列中已经存在的元素总数。
1.3 Condition实现
Condition 现在的实现只有java.util.concurrent.locks.AbstractQueueSynchoronizer内部的 ConditionObject,并且通过ReentranLock的newCondition()方法暴露出来,这是因为Condition的 await()/sinal()一般在lock.lock()与lock.unlock()之间执行,当执行condition.await()方法时, 它会首先释放掉本线程持有的锁,然后自己进入等待队列。直到sinal(),唤醒后又会重新试图去拿到锁,拿到后执行await()下的代码,其中释放当 前锁和得到当前锁都需要ReentranLock的tryAcquire(int arg)方法来判定,并且享受ReentranLock的重进入特性。
Java代码
- public final void await() throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- //加一个新的condition等待节点
- Node node = addConditionWaiter();
- //释放自己的锁
- int savedState = fullyRelease(node);
- int interruptMode = 0;
- while (!isOnSyncQueue(node)) {
- //如果当前线程 等待状态时CONDITION,park住当前线程,等待condition的signal来解除
- LockSupport.park(this);
- if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
- break;
- }
- if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
- interruptMode = REINTERRUPT;
- if (node.nextWaiter != null)
- unlinkCancelledWaiters();
- if (interruptMode != 0)
- reportInterruptAfterWait(interruptMode);
- }
2、SynchronousQueue
一种阻塞队列,其中每个 put 必须等待一个 take,反之亦然。同步队列没有任何内部容量,甚至连一个队列的容量都没有。不能在同步队列上进行 peek,因为仅在试图要取得元素时,该元素才存在;除非另一个线程试图移除某个元素,否则也不能(使用任何方法)添加元素;也不能迭代队列,因为其中没有元素可用于迭代。队列的头 是尝试添加到队列中的首个已排队线程元素;如果没有已排队线程,则不添加元素并且头为 null。对于其他Collection 方法(例如 contains),SynchronousQueue 作为一个空集合。此队列不允许 null 元素。
同步队列类似于 CSP 和 Ada 中使用的 rendezvous 信道。它非常适合于传递性设计,在这种设计中,在一个线程中运行的对象要将某些信息、事件或任务传递给在另一个线程中运行的对象,它就必须与该对象同步。
对于正在等待的生产者和使用者线程而言,此类支持可选的公平排序策略。默认情况下不保证这种排序。但是,使用公平设置为 true 所构造的队列可保证线程以 FIFO 的顺序进行访问。公平通常会降低吞吐量,但是可以减小可变性并避免得不到服务。
3、LinkedBlockingQueue
单 向链表结构的队列。如果不指定容量默认为Integer.MAX_VALUE。通过putLock和takeLock两个锁进行同步,两个锁分别实例化 notFull和notEmpty两个Condtion,用来协调多线程的存取动作。其中某些方法(如 remove,toArray,toString,clear等)的同步需要同时获得这两个锁,并且总是先putLock.lock紧接着 takeLock.lock(在同一方法fullyLock中),这样的顺序是为了避免可能出现的死锁情况(我也想不明白为什么会是这样?)
4、PriorityBlockingQueue
PriorityQueue
相同的顺序规则,并且提供了阻塞检索的操作。虽然此队列逻辑上是无界的,但是由于资源被耗尽,所以试图执行添加操作可能会失败(导致 OutOfMemoryError)。此类不允许使用 null 元素。依赖自然顺序的优先级队列也不允许插入不可比较的对象(因为这样做会抛出ClassCastException)。看它的三个属性,就基本能看懂这个类了:
Java代码
- private final PriorityQueue q;
- private final ReentrantLock lock = new ReentrantLock(true);
- private final Condition notEmpty = lock.newCondition();
说明,本类内部数据结构是PriorityQueue,至于PriorityQueue怎么排序看之前一篇文章:http://jiadongkai-sina-com.iteye.com/blog/825683
lock说明本类使用一个lock来同步读写等操作。
notEmpty协调队列是否有新元素提供,而队列满了以后会调用PriorityQueue的grow方法来扩容。
5、DelayQueue
Delayed接口继承自Comparable,我们插入的E元素都要实现这个接口。
DelayQueue的设计目的间API文档:
An unbounded blocking queue of Delayed elements, in which an element can only be taken when its delay has expired. The head of the queue is that Delayed element whose delay expired furthest in the past. If no delay has expired there is no head and poll will returnnull. Expiration occurs when an element's getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero. Even though unexpired elements cannot be removed using take or poll, they are otherwise treated as normal elements. For example, the size method returns the count of both expired and unexpired elements. This queue does not permit null elements.
因 为DelayQueue构造函数了里限定死不允许传入comparator(之前的PriorityBlockingQueue中没有限定死),即只能在 compare方法里定义优先级的比较规则。再看上面这段英文,“The head of the queue is that Delayed element whose delay expired furthest in the past.”说明compare方法实现的时候要保证最先加入的元素最早结束延时。而 “Expiration occurs when an element's getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero.”说明getDelay方法的实现必须保证延时到了返回的值变为<=0的int。
上 面这段英文中,还说明了:在poll/take的时候,队列中元素会判定这个elment有没有达到超时时间,如果没有达到,poll返回null,而 take进入等待状态。但是,除了这两个方法,队列中的元素会被当做正常的元素来对待。例如,size方法返回所有元素的数量,而不管它们有没有达到超时 时间。而协调的Condition available只对take和poll是有意义的。
另 外需要补充的是,在ScheduledThreadPoolExecutor中工作队列类型是它的内部类DelayedWorkQueue,而 DelayedWorkQueue的Task容器是DelayQueue类型,而ScheduledFutureTask作为Delay的实现类作为 Runnable的封装后的Task类。也就是说ScheduledThreadPoolExecutor是通过DelayQueue优先级判定规则来执 行任务的。
6、BlockingDque+LinkedBlockingQueue
BlockingDque 为阻塞双端队列接口,实现类有LinkedBlockingDque。双端队列特别之处是它首尾都可以操作。LinkedBlockingDque不同于 LinkedBlockingQueue,它只用一个lock来维护读写操作,并由这个lock实例化出两个Condition notEmpty及notFull,而LinkedBlockingQueue读和写分别维护一个lock。
Java线程(十三):BlockingQueue-线程的阻塞队列
http://blog.csdn.net/zzp_403184692/article/details/8021615
(13)多线程与并发库之java5阻塞队列(BlockingQueue)的应用----子线程循环10次,接着主线程循环100次,接着又回到子线程循环10次,接着再回到主线程循环100次,如此循环50次
http://blog.csdn.net/itm_hadf/article/details/7538083
线程----BlockingQueue (转)
http://www.cnblogs.com/likwo/archive/2010/07/01/1769199.html
并发容器——BlockingQueue相关类
http://developer.51cto.com/art/201104/256805.htm
Java Concurrent包学习之BlockingQueue
http://blog.csdn.net/derekjiang/article/details/5330019
相关推荐
Java多线程之BlockingQueue深入分析 Java多线程中,BlockingQueue是一种特殊的队列,它可以为线程同步提供有力的保障。在Java多线程(五)之BlockingQueue深入分析中,我们将深入分析BlockingQueue的定义、常用方法...
本文将深入探讨BlockingQueue的使用、特性以及常见操作。 首先, BlockingQueue接口位于`java.util.concurrent`包下,它继承自`java.util.Queue`接口,并添加了一些阻塞操作。阻塞操作指的是当队列为空时,获取元素...
阻塞队列BlockingQueue是Java并发编程中一个重要的数据结构,它是线程安全的队列,主要用于生产者消费者模型中的数据交换。在Java的`java.util.concurrent`包中,提供了多种实现阻塞队列的类,如`ArrayBlockingQueue...
本文将深入探讨 BlockingQueue 的原理、特性以及如何在实际应用中使用。 ### 1. BlockingQueue概述 BlockingQueue 是一个并发容器,它实现了队列的接口,同时具备阻塞功能。当队列为空时,尝试从队列中获取元素的...
通过阅读和分析这些代码,你可以深入理解如何在Spring MVC项目中集成和利用`BlockingQueue`和`@PostConstruct`注解。 总结来说,`spring MVC 初始启动concurrent blocking queue`涉及的是在Spring MVC应用启动时...
### 三、 blockingqueue-example 示例项目分析 在 "blockingqueue-example-master" 示例项目中,我们可以看到如何使用 BlockingQueue 进行多线程交互。项目通常包含生产者(Producer)和消费者(Consumer)两个线程...
本实例提供了源代码和详细指导教程,帮助开发者深入理解这一概念。 生产者消费者问题通常使用Java中的`java.util.concurrent`包来解决,特别是`BlockingQueue`接口。在这个例子中,`BlockingQueue`作为共享的缓冲区...
通过分析这些代码,可以更深入地理解多线程编程和Java并发机制。学习和理解这个问题对于进行高并发系统设计和优化是非常重要的,因为它是并发编程中常见的模式,广泛应用于消息队列、数据库连接池等场景。
通过分析和实践`BlockingQueueDemo`,开发者可以深入理解Android中消息队列的工作原理,提高应用程序的响应性和可维护性。这种自定义的消息队列设计对于实现复杂的定时任务、动画控制或者其他需要精细控制的任务调度...
本文将深入探讨Java多线程编程的基础知识,包括线程的创建、同步、通信以及相关实例。 一、线程的创建 在Java中,有两种主要的创建线程的方式: 1. 继承Thread类:创建一个新的类,继承自Thread类,然后重写它的...
通过这个实验,你可以深入理解Java并发编程中的同步与异步概念,以及`BlockingQueue`如何在生产者消费者问题中起到关键作用。此外,你还可以学习到如何在实际项目中使用Java的并发工具来提高多线程环境下的程序性能...
在本练习中,我们将通过`BlockingQueue1.java`和`BlockingQueue2.java`两个文件来深入理解和实践BlockQueue的基本用法和实现。 BlockQueue的核心特性在于它的阻塞操作:当队列为空时,尝试获取元素的线程会被阻塞;...
首先,我们深入理解生产者消费者模型。该模型基于“缓冲区”概念,生产者将产品放入缓冲区,而消费者从缓冲区取出产品。当缓冲区满时,生产者需要等待;当缓冲区空时,消费者也需要等待。这种机制由Java的并发工具类...
在源码层面,我们可以通过分析`BlockingQueue`接口的实现,如`ArrayBlockingQueue`、`LinkedBlockingQueue`等,来了解其内部是如何处理线程的阻塞和唤醒的。这些实现通常会用到`wait()`和`notify()`方法,或者`Lock`...
《生产者-消费者问题的深度剖析与探讨》 在计算机科学和并发编程中,"生产者-消费者"问题是一个经典的问题模型,它涉及到多...通过对提供的代码文件的分析,我们可以深入理解这一模式在实际编程中的运用和实现细节。
通过对这些代码的学习和分析,可以帮助开发者深入理解Java模拟Spooling的具体实现方法。 通过以上讲解,我们可以看到Java虽然没有内置的Spooling机制,但借助其强大的并发库和数据结构,完全可以实现一个高效的假...
生产者消费者问题是多线程编程中的一个经典案例,主要展示了线程间的同步和通信。...通过分析源码,我们可以深入理解如何利用`BlockingQueue`等工具解决这类问题,并学习到如何在多线程环境中确保程序的正确性和性能。
- **类库结构**:J.U.C提供了丰富的类库支持,包括但不限于`ExecutorService`、`Future`、`BlockingQueue`、`CountDownLatch`、`Semaphore`、`CyclicBarrier`等。 - **样例分析**:通过具体的代码示例来讲解这些类...
Java集合框架源码分析 Java集合框架是Java语言中一个非常重要的组件,提供了多种数据结构和算法来存储和操作数据。在Java集合框架中,...本文对Java集合框架的源码进行了深入分析,为读者提供了详细的参考资料。
在这个"模仿线程"生产者与消费者"的例子中,我们将深入探讨这个经典的并发设计模式及其背后的原理。 生产者-消费者模式是一种典型的同步问题,用于解决数据生产与消费的异步处理。在这个模型中,生产者线程负责创建...