JAVA 的BlockingQueue接口, java.util.concurrent.BlockingQueue, 代表着一个队列可以安全的插入和取元素.换句话说,多线程通过BlockingQueue安全的插入或者取元素,不会有任何并发问题发生。
BlockingQueue用法
BlockingQueue典型的应用是一个生产者多个消费者,下面一张图描述了其原理:
一个线程放入元素到BlockingQueue,其他线程从中取元素
生产者不断的生产新对象并且加到队列BlockingQueue中,直到队列加满,也就是说如果队列满了,那么生产者就会一直阻塞,直到消费者取出元素。
消费者不断的从BlockingQueue 中取元素,如果队列为空,那么一直阻塞直到生产者往队列中加入元素。
BlockingQueue的方法
BlockingQueue接口中分别有4种方法不同的插入和移除元素的方法,已经两种检查元素的方法。如果请求的操作无法执行,则每组方法的行为不同。
下面的表格是4种方法:
下面是4种方法的解释:
Throws Exception:
尝试操作失败,则会抛出异常
Special Value:
尝试操作,成功会返回true或者失败返回false
Blocks:
如果不能立马操作,则会阻塞,直到可以操作
Times Out:
尝试操作,如果不能立马操作,则会在指定的时间内返回成功true或者失败false.
不能在BlockingQueue中加入null,否则会抛NullPointerException。
可以访问BlockingQueue中的所有元素,而不仅仅是头和尾。例如,假设您已将一个对象排队等待处理,但您的应用程序决定取消它。然后可以调用remove(o)来删除队列中的特定对象。但是,这并不是很有效,所以除非您真的需要,否则不应该使用这些 Collection方法.
BlockingQueue的实现:
BlockingQueue是个接口,所以需要用它的实现类, java.util.concurrent 包下实现了BlockingQueue的类:
ArrayBlockingQueue
DelayQueue
LinkedBlockingQueue
PriorityBlockingQueue
SynchronousQueue
这些实现类我们后面会一一讲解。
BlockingQueue的简单例子:
这儿用了 BlockingQueue 的实现类ArrayBlockingQueue .
首先在BlockingQueueExample 分别启动生产者和消费者线程,生产者往BlockingQueue 中插入String元素,消费者从中取出。
public class BlockingQueueExample {
public static void main(String[] args) throws Exception {
BlockingQueue queue = new ArrayBlockingQueue(1024);
Producer producer = new Producer(queue);
Consumer consumer = new Consumer(queue);
new Thread(producer).start();
new Thread(consumer).start();
Thread.sleep(4000);
}
}
这是生产者类,注意每次put()之前都休眠了1s,这回使得消费者阻塞,等待往队列里面加入元素:
public class Producer implements Runnable{
protected BlockingQueue queue = null;
public Producer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
try {
queue.put("1");
Thread.sleep(1000);
queue.put("2");
Thread.sleep(1000);
queue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
这是消费者类,就是从队列中取元素打印:.
public class Consumer implements Runnable{
protected BlockingQueue queue = null;
public Consumer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
try {
System.out.println(queue.take());
System.out.println(queue.take());
System.out.println(queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}