个人中心
文章发布
退出
作者热门文章
- ubuntu12.04环境下使用kvm ioctl接口实现最简单的虚拟机
- Ubuntu 通过无线网络安装Ubuntu Server启动系统后连接无线网络的方法
- 在Ubuntu上搭建网桥的方法
- ubuntu 虚拟机上网方式及相关配置详解
25
4
CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.
这篇CFSDN的博客文章Java并发之BlockingQueue的使用由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.
本文主要讲的是并发包中涉及到的集合，关于普通集合，请参考【java 集合概览】 。
1、什么是blockingqueue 。
blockingqueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种:
1. 当队列满了的时候进行入队列操作 2. 当队列空了的时候进行出队列操作 。
因此，当一个线程试图对一个已经满了的队列进行入队列操作时，它将会被阻塞，除非有另一个线程做了出队列操作；同样，当一个线程试图对一个空队列进行出队列操作时，它将会被阻塞，除非有另一个线程进行了入队列操作.
在java中，blockingqueue的接口位于java.util.concurrent 包中(在java5版本开始提供)，由上面介绍的阻塞队列的特性可知，阻塞队列是线程安全的.
2、blockingqueue的用法 。
阻塞队列主要用在生产者/消费者的场景，下面这幅图展示了一个线程生产、一个线程消费的场景:
负责生产的线程不断的制造新对象并插入到阻塞队列中，直到达到这个队列的上限值。队列达到上限值之后生产线程将会被阻塞，直到消费的线程对这个队列进行消费。同理，负责消费的线程不断的从队列中消费对象，直到这个队列为空，当队列为空时，消费线程将会被阻塞，除非队列中有新的对象被插入.
3、blockingqueue接口中的方法 。
阻塞队列一共有四套方法分别用来进行insert、remove和examine，当每套方法对应的操作不能马上执行时会有不同的反应，下面这个表格就分类列出了这些方法:
。
| - | throws exception | special value | blocks | times out |
|---|---|---|---|---|
| insert | add(o) | offer(o) | put(o) | offer(o, timeout, timeunit) |
| remove | remove(o) | poll() | take() | poll(timeout, timeunit) |
| examine | element() | peek() |
。
这四套方法对应的特点分别是:
1. throwsexception：如果操作不能马上进行，则抛出异常 2. specialvalue：如果操作不能马上进行，将会返回一个特殊的值，一般是true或者false 3. blocks:如果操作不能马上进行，操作会被阻塞 4. timesout:如果操作不能马上进行，操作会被阻塞指定的时间，如果指定时间没执行，则返回一个特殊值，一般是true或者false 。
需要注意的是，我们不能向blockingqueue中插入null，否则会报nullpointerexception.
4、blockingqueue的实现类 。
blockingqueue只是java.util.concurrent包中的一个接口，而在具体使用时，我们用到的是它的实现类，当然这些实现类也位于java.util.concurrent包中。在java6中，blockingqueue的实现类主要有以下几种:
1. arrayblockingqueue 2. delayqueue 3. linkedblockingqueue 4. priorityblockingqueue 5. synchronousqueue 。
下面我们就分别介绍这几个实现类.
4.1 arrayblockingqueue 。
arrayblockingqueue是一个有边界的阻塞队列，它的内部实现是一个数组。有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变.
arrayblockingqueue是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。下面是一个初始化和使用arrayblockingqueue的例子:
|
1
2
3
|
blockingqueue queue =
new
arrayblockingqueue(
1024
);
queue.put(
"1"
);
object object = queue.take();
|
4.2 delayqueue 。
delayqueue阻塞的是其内部元素，delayqueue中的元素必须实现 java.util.concurrent.delayed接口，这个接口的定义非常简单:
|
1
2
3
|
public
interface
delayed
extends
comparable<delayed> {
long
getdelay(timeunit unit);
}
|
getdelay()方法的返回值就是队列元素被释放前的保持时间，如果返回0或者一个负值，就意味着该元素已经到期需要被释放，此时delayedqueue会通过其take()方法释放此对象.
从上面delayed 接口定义可以看到，它还继承了comparable接口，这是因为delayedqueue中的元素需要进行排序，一般情况，我们都是按元素过期时间的优先级进行排序.
例1：为一个对象指定过期时间 。
首先，我们先定义一个元素，这个元素要实现delayed接口 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
public
class
delayedelement
implements
delayed {
private
long
expired;
private
long
delay;
private
string name;
delayedelement(string elementname,
long
delay) {
this
. name = elementname;
this
. delay= delay;
expired = ( delay + system. currenttimemillis());
}
@override
public
int
compareto(delayed o) {
delayedelement cached=(delayedelement) o;
return
cached.getexpired()> expired?
1
:-
1
;
}
@override
public
long
getdelay(timeunit unit) {
return
( expired - system. currenttimemillis());
}
@override
public
string tostring() {
return
"delayedelement [delay="
+ delay +
", name="
+ name +
"]"
;
}
public
long
getexpired() {
return
expired;
}
}
|
设置这个元素的过期时间为3s 。
|
1
2
3
4
5
6
7
8
9
|
public
class
delayqueueexample {
public
static
void
main(string[] args)
throws
interruptedexception {
delayqueue<delayedelement> queue=
new
delayqueue<>();
delayedelement ele=
new
delayedelement(
"cache 3 seconds"
,
3000
);
queue.put( ele);
system. out.println( queue.take());
}
}
|
运行这个main函数，我们可以发现，我们需要等待3s之后才会打印这个对象.
其实delayqueue应用场景很多，比如定时关闭连接、缓存对象，超时处理等各种场景，下面我们就拿学生考试为例让大家更深入的理解delayqueue的使用.
例2：把所有考试的学生看做是一个delayqueue，谁先做完题目释放谁 。
首先，我们构造一个学生对象 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
public
class
student
implements
runnable,delayed{
private
string name;
//姓名
private
long
costtime;
//做试题的时间
private
long
finishedtime;
//完成时间
public
student(string name,
long
costtime) {
this
. name = name;
this
. costtime= costtime;
finishedtime = costtime + system. currenttimemillis();
}
@override
public
void
run() {
system. out.println( name +
" 交卷,用时"
+ costtime /
1000
);
}
@override
public
long
getdelay(timeunit unit) {
return
( finishedtime - system. currenttimemillis());
}
@override
public
int
compareto(delayed o) {
student other = (student) o;
return
costtime >= other. costtime?
1
:-
1
;
}
}
|
然后在构造一个教师对象对学生进行考试 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
public
class
teacher {
static
final
int
student_size =
30
;
public
static
void
main(string[] args)
throws
interruptedexception {
random r =
new
random();
//把所有学生看做一个延迟队列
delayqueue<student> students =
new
delayqueue<student>();
//构造一个线程池用来让学生们“做作业”
executorservice exec = executors.newfixedthreadpool(student_size);
for
(
int
i =
0
; i < student_size; i++) {
//初始化学生的姓名和做题时间
students.put(
new
student(
"学生"
+ (i +
1
),
3000
+ r.nextint(
10000
)));
}
//开始做题
while
(! students.isempty()){
exec.execute( students.take());
}
exec.shutdown();
}
}
|
我们看一下运行结果:
学生2 交卷,用时3 学生1 交卷,用时5 学生5 交卷,用时7 学生4 交卷,用时8 学生3 交卷,用时11 。
通过运行结果我们可以发现，每个学生在指定开始时间到达之后就会“交卷”（取决于getdelay()方法），并且是先做完的先交卷（取决于compareto()方法）.
通过查看其源码可以看到，delayqueue内部实现用的是priorityqueue和一个lock:
4.3 linkedblockingqueue 。
linkedblockingqueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为integer.max_value的容量 。它的内部实现是一个链表.
和arrayblockingqueue一样，linkedblockingqueue 也是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。下面是一个初始化和使linkedblockingqueue的例子:
|
1
2
3
4
|
blockingqueue<string> unbounded =
new
linkedblockingqueue<string>();
blockingqueue<string> bounded =
new
linkedblockingqueue<string>(
1024
);
bounded.put(
"value"
);
string value = bounded.take();
|
4.4 priorityblockingqueue 。
priorityblockingqueue是一个没有边界的队列，它的排序规则和 java.util.priorityqueue一样。需要注意，priorityblockingqueue中允许插入null对象.
所有插入priorityblockingqueue的对象必须实现 java.lang.comparable接口，队列优先级的排序规则就是按照我们对这个接口的实现来定义的.
另外，我们可以从priorityblockingqueue获得一个迭代器iterator，但这个迭代器并不保证按照优先级顺序进行迭代.
下面我们举个例子来说明一下，首先我们定义一个对象类型，这个对象需要实现comparable接口:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
public
class
priorityelement
implements
comparable<priorityelement> {
private
int
priority;
//定义优先级
priorityelement(
int
priority) {
//初始化优先级
this
.priority = priority;
}
@override
public
int
compareto(priorityelement o) {
//按照优先级大小进行排序
return
priority >= o.getpriority() ?
1
: -
1
;
}
public
int
getpriority() {
return
priority;
}
public
void
setpriority(
int
priority) {
this
.priority = priority;
}
@override
public
string tostring() {
return
"priorityelement [priority="
+ priority +
"]"
;
}
}
|
然后我们把这些元素随机设置优先级放入队列中 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
public
class
priorityblockingqueueexample {
public
static
void
main(string[] args)
throws
interruptedexception {
priorityblockingqueue<priorityelement> queue =
new
priorityblockingqueue<>();
for
(
int
i =
0
; i <
5
; i++) {
random random=
new
random();
priorityelement ele =
new
priorityelement(random.nextint(
10
));
queue.put(ele);
}
while
(!queue.isempty()){
system.out.println(queue.take());
}
}
}
|
看一下运行结果:
priorityelement [priority=3] priorityelement [priority=4] priorityelement [priority=5] priorityelement [priority=8] priorityelement [priority=9] 。
4.5 synchronousqueue 。
synchronousqueue队列内部仅允许容纳一个元素。当一个线程插入一个元素后会被阻塞，除非这个元素被另一个线程消费.
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持我.
原文链接：https://blog.csdn.net/suifeng3051/article/details/48807423 。
最后此篇关于Java并发之BlockingQueue的使用的文章就讲到这里了,如果你想了解更多关于Java并发之BlockingQueue的使用的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。
25
4
0
我的理解是1.5加入了callable，runnable接口(interface)保持原样，防止世界末日。为什么我不能实例化 ThreadPoolExecutor (core, max, tu, un
我正在制作一个包含两个线程的应用程序:其中一个向 LinkedBlockingQueue 写入一个值，另一个正在读取。我正在使用 ScheduledExecutorService 在某个时间段内以秒为
我正在尝试 Autowiring 参数化的阻塞队列: @Bean(name = "saveProductQueue") public BlockingQueue saveProductQueue()
我正在实现一个程序，其中主线程将各种消息推送到工作线程，而工作线程将工作结果推送回主线程。 为此，我计划使用两个队列，一个用于推送到工作线程，另一个用于从中拉出。 据我了解，线程会缓存对象，因此如果它
我正在使用 RabbitMQ，它默认为消费者使用 LinkedBlockingQueue 。它有一个阻塞的 nextDelivery() 方法，该方法基本上调用队列上的 take() 。 但是如果在调
我有 n 个生产者线程通过 BlockingQueue 为 1 个消费者线程提供数据。我正在使用 .put 和 .take (后者当 .peek != null 时)。除了明显在传输过程中不可避免的数
我已经实现了一个带有套接字线程池的两人游戏。每个玩家都连接到自己的线程。我按照this添加了一个消息队列系统文章。 问题是消息滞后。第一个玩家的第一个响应会按预期添加到 messageQueue 中。
据我所知，BlockingCollection 使用非繁忙等待，这是对新项目/回调的通知。所以我不明白它是如何阻塞的，但我认为我可能混合了阻塞的线程和阻塞的共享对象访问？ 最佳答案 这是一个很好的解释
我的 Android 应用程序有一个长时间运行的后台服务，据我所知，它在应用程序的主线程中运行，因此，任何耗时或阻塞的任务都应移至单独的线程。 现在，情况是这样的，我不明白/困惑: 当我从一个 Act
我在数据库前面使用 LinkedBlockingQueue。一个线程写入队列，另一个线程从队列读取。 我认为两个并发写入是不可能的。但是是否有可能一个线程写入而另一个线程同时从队列中读取呢？如果没有，
我有一个在单个后台线程上处理工作事件的 BlockingQueue。各种线程调用 add 将一些工作添加到队列中，单个后台线程调用 take 获取工作并一次处理一个。最终可能是停止处理工作的时候了，我
我在多线程系统中使用 BlockingQueue，其中同步块(synchronized block)将项目添加到列表中。有时它不会将项目添加到列表中，它遗漏的项目是随机的。我尝试将以下行添加到代码中，
我有以下阻塞队列； final BlockingQueue blockingQueue = new LinkedBlockingQueue(); 在哪里 public class Message
这个问题的标题让我怀疑这是否存在，但仍然: 我感兴趣的是是否有 Java 的 BlockingQueue 的实现，它受大小限制，从不阻塞，而是在尝试入队太多元素时抛出异常。 编辑 - 我将 Block
假设我有 BlockingQueue 并且一些线程被称为 take() 但此时队列是空的。假设我以某种方式知道将来不会有新元素出现在队列中。如何释放那些被称为 take() 的线程等待？谢谢! p
我想要一个线程安全的容器，它会阻止调用者，直到有项目可用为止。项目将以每秒 1000 秒的速度添加到此容器中，但不会以相同的速度排出。因此，我希望容器不允许重复。我围绕 LinkedBlockingQ
我正在使用 BlockingQueue(LinkedBlockingQueue) 在多个线程之间同步数据。请看下图。 主线程是一个生产者，它生产对象，然后将它们放入每个消费者的队列中(线程2-10)。
我有一个容量为 1 的 BlockingQueue。它存储收到的股票的最后价格。价格保留在队列中，直到客户端轮询队列。然后，我有一个名为 getLatestPrice() 的方法，它应该返回股票的最新
我的 Java 代码中使用 BlockingQueue 时存在潜在的竞争条件，我想知道如何修改代码来避免这种情况: private static BlockingQueue ftpQueue = ne
我有一种情况需要 BlockingQueue 上的方法 peekWait。我会将此方法描述为 retrieves but not remove the head of the queue, waiti
我是一名优秀的程序员,十分优秀！