个人中心
gpt4 book ai didi

Java并发之BlockingQueue的使用

转载 作者：qq735679552 更新时间：2022-09-28 22:32:09 25 4
gpt4 key购买 nike

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章Java并发之BlockingQueue的使用由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

本文主要讲的是并发包中涉及到的集合，关于普通集合，请参考【java 集合概览】 。

1、什么是blockingqueue 。

blockingqueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种:

1. 当队列满了的时候进行入队列操作 2. 当队列空了的时候进行出队列操作 。

因此，当一个线程试图对一个已经满了的队列进行入队列操作时，它将会被阻塞，除非有另一个线程做了出队列操作；同样，当一个线程试图对一个空队列进行出队列操作时，它将会被阻塞，除非有另一个线程进行了入队列操作.

在java中，blockingqueue的接口位于java.util.concurrent 包中(在java5版本开始提供)，由上面介绍的阻塞队列的特性可知，阻塞队列是线程安全的.

2、blockingqueue的用法 。

阻塞队列主要用在生产者/消费者的场景，下面这幅图展示了一个线程生产、一个线程消费的场景:

Java并发之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:

Java并发之BlockingQueue的使用

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
文章推荐： mysql升级到5.7时,wordpress导数据报错1067的问题
文章推荐： 10月份GitHub上热门开源项目排行
文章推荐： 解决Navicat for Mysql连接报错1251的问题(连接失败)
文章推荐： VS Code 还是 Atom？你应该选择哪个
qq735679552
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
全站热门文章
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com