Java java.util.currency
包中的 BlockingQueue
接口意味着线程可以在一个队列中添加和提取对象.在接下来的段落中,我们将学习如何使用 BlockingQueue
接口.
BlockingQueue典型应用是用于一个线程生产对象,另一个线程消费对象.
下面是一个原理描述图:
一个阻塞队列伴随着一个线程往队列中添加对象,另一个线程从队列中提取对象.
生产线程负责生产新对象并发它添加进队列中,直到达到队列所能容纳的边界值为止.即达到容量限制.如果队列达到容量限制,则生产线程在插入新对象时将会发生阻塞.直到有一个消费线程从队列中取走一个对象为止.
消费线程负责从阻塞队列中取出对象并处理它们.如果消费线程尝试从空队列中取出空队列,那么消费线程将会阻塞直到生产线程往队列中放入一个对象为止.
一个阻塞队列包含四种截然不同的方法集,用于完成在队列中插入,移除和检查元素.每种集合的不同之处在于操作请求不能被立即执行时的不同行为.下面是这些方法的列表:
Throws Exception | Special Value | Blocks | Times Out | |
---|---|---|---|---|
Insert | add(o) | offer(o) | put(o) | offer(o, timeout, time unit) |
Remove | remove(o) | poll() | take() | poll(timeout, timeunit) |
Examine | element() | peek() |
四种不同的行为结合如下:
不能插入 null
到BlockingQueue中.如果你这么做了,BlockingQueue将抛出一个 NullPointerException
.
当然我们也能够访问BlockingQueue的所有元素, 不只是队头和队尾. 例如, 你需要出队一个对象用于执行,但你的应用决定取消它.那么你仍然可以使用类似remove(o)方法来移除队列中指定的对象.然而这不是很高效,所以你不应该使用这类方法, 除非你真的需要这么做.
BlockingQueue只是一个接口, 所以你需要使用它众多实现的其中一种来使用它. java.util.concurrency
包中有如下几种BlockingQueue的接口实现(在Java6中):
下面是一个Java BlockingQueue的实例. 实例中使用BlockingQueue接口的实现类,ArrayBlockingQueue.
首先,BlockingQueueExample类中分别启动一个生产和消费线程.生产线程负责插入字符串到共享BlockingQueue中,而消费线程则负责取走它们.
public class ArrayBlockingQueueExample { public static void main(String[] args) { ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(1024); Producer producer = new Producer(queue); Consumer consumer = new Consumer(queue); new Thread(producer).start(); new Thread(consumer).start(); try { Thread.sleep(4000); } catch (InterruptedException e) { e.printStackTrace(); } } } 复制代码
下面是Producer类.注意我们每调用一次put()就会让线程睡眠一秒.这会导致消费线程在队列中等待对象时发生阻塞.
public class Producer implements Runnable { private ArrayBlockingQueue<String> queue; public Producer(ArrayBlockingQueue<String> queue) { this.queue = queue; } @Override public void run() { try { queue.put("1"); Thread.sleep(1000); queue.put("2"); Thread.sleep(1000); queue.put("3"); Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } 复制代码
下面是Consumer类.它只是负责从队列中取出字符串并通过System.out打印出来.
package org.menfre; import java.util.concurrent.ArrayBlockingQueue; public class Consumer implements Runnable { private ArrayBlockingQueue<String> queue; public Consumer(ArrayBlockingQueue<String> queue){ this.queue = queue; } @Override public void run() { try { System.out.println(queue.take()); System.out.println(queue.take()); System.out.println(queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } } } 复制代码
该系列博文为笔者复习基础所著译文或理解后的产物,复习原文来自Jakob Jenkov所著Java.util.concurrent
Next: ArrayBlockingQueue