欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Java并发之BlockingQueue的使用

程序员文章站 2023-12-19 13:37:40
本文主要讲的是并发包中涉及到的集合,关于普通集合,请参考【】 一、什么是blockingqueue blockingqueue即阻塞队列,从阻塞这个词可以看出,在某些情...

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

一、什么是blockingqueue

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

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

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

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

二、blockingqueue的用法

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

Java并发之BlockingQueue的使用

负责生产的线程不断的制造新对象并插入到阻塞队列中,直到达到这个队列的上限值。队列达到上限值之后生产线程将会被阻塞,直到消费的线程对这个队列进行消费。同理,负责消费的线程不断的从队列中消费对象,直到这个队列为空,当队列为空时,消费线程将会被阻塞,除非队列中有新的对象被插入。

三、blockingqueue接口中的方法

阻塞队列一共有四套方法分别用来进行insertremoveexamine,当每套方法对应的操作不能马上执行时会有不同的反应,下面这个表格就分类列出了这些方法:

- 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

四、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的例子:

blockingqueue queue = new arrayblockingqueue(1024);
queue.put("1");
object object = queue.take();

4.2 delayqueue

delayqueue阻塞的是其内部元素,delayqueue中的元素必须实现 java.util.concurrent.delayed接口,这个接口的定义非常简单:

public interface delayed extends comparable<delayed> {
long getdelay(timeunit unit);
}

getdelay()方法的返回值就是队列元素被释放前的保持时间,如果返回0或者一个负值,就意味着该元素已经到期需要被释放,此时delayedqueue会通过其take()方法释放此对象。

从上面delayed 接口定义可以看到,它还继承了comparable接口,这是因为delayedqueue中的元素需要进行排序,一般情况,我们都是按元素过期时间的优先级进行排序。

例1:为一个对象指定过期时间

首先,我们先定义一个元素,这个元素要实现delayed接口

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

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,谁先做完题目释放谁

首先,我们构造一个学生对象

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;
 }

}

然后在构造一个教师对象对学生进行考试

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的例子:

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接口:

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 + "]";
}
}

然后我们把这些元素随机设置优先级放入队列中

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队列内部仅允许容纳一个元素。当一个线程插入一个元素后会被阻塞,除非这个元素被另一个线程消费。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。

上一篇:

下一篇: