阻塞队列是在队列基础上又加了两个附加操作的队列,两个操作时阻塞插入和阻塞移除。
阻塞插入:队列满时,队列会阻塞插入元素的线程,直到队列不满
阻塞移除:队列空时,获取元素的线程会等待队列变为非空。
一个简单例子:
public class Thread_Bdueue {
public static void main(String args[]){
BlockingDeque blockingDeque=new LinkedBlockingDeque(20);
for (int i=0;i<30;i++){
try {
//指定元素添加到阻塞栈当中,如果没有可用空间,则一直等待
blockingDeque.putFirst(i);
System.out.println("向阻塞栈当中添加了元素:"+i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("程序到此运行结束-------");
}
}
应用场景:阻塞队列应用于生产者消费者的场景,简而言之,阻塞队列是生产者存放元素,消费者消费元素的容器。
方法\处理方式 | 抛出异常 | 返回特殊值 | 一直阻塞 | 超时退出 |
---|---|---|---|---|
插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
移除方法 | remove() | poll() | take() | poll(time,unit) |
检查方法 | element() | peek() | 不可用 | 不可用 |
JDK7提供了7个阻塞队列
(1)、ArrayBlockingQueue :一个由数组结构组成的有界阻塞队列。
此队列按照先进先出的原色对元素进行排序,默认情况下不保证公平的访问队列。即如果队列满了,那么被阻塞在外面的线程对队列访问的顺序是不能保证线程公平(即先阻塞,先插入)的。
(2)、LinkedBlockingQueue :一个由链表结构组成的有界阻塞队列。此队列按照先出先进的原则对元素进行排序
(3)、PriorityBlockingQueue :一个支持优先级排序的无界阻塞队列。
(4)、DelayQueue:一个使用优先级队列实现的无界阻塞队列。
(5)、SynchronousQueue:一个不存储元素的阻塞队列。每一个put必须等待一个take操作,否则不能继续添加元素。并且他支持公平访问队列。
(6)、LinkedTransferQueue:一个由链表结构组成的无界阻塞队列。
(7)、LinkedBlockingDeque:一个由链表结构组成的双向阻塞队列。优势在于多线程入队时,减少一半的竞争。
使用BlockingQueue解决生产者和消费者的问题:
任何有效的生产者-消费者问题解决方案都是通过控制生产者put()方法(生产资源)和消费者take()方法(消费资源)的调用来实现的,一旦你实现了对方法的阻塞控制,那么你将解决该问题。
Java通过BlockingQueue提供了开箱即用的支持来控制这些方法的调用(一个线程创建资源,另一个消费资源)。java.util.concurrent包下的BlockingQueue接口是一个线程安全的可用于存取对象的队列。
生产者:
public class Producer implements Runnable {
protected BlockingQueue<Object> queue;
Producer(BlockingQueue<Object> theQueue) {
this.queue = theQueue;
}
Object getResource() {
try {
Thread.sleep(100);
} catch (InterruptedException ex) {
System.out.println("生产者读中断");
}
return new Object();
}
@Override
public void run() {
try {
while (true) {
Object justProduced = getResource();
queue.put(justProduced);
System.out.println("生产者资源队列大小= " + queue.size());
}
} catch (InterruptedException ex) {
System.out.println("生产者中断");
}
}
}
消费者:
public class Consumer implements Runnable {
protected BlockingQueue<Object> queue;
Consumer(BlockingQueue<Object> theQueue) {
this.queue = theQueue;
}
void take(Object obj) {
try {
Thread.sleep(100); // simulate time passing
} catch (InterruptedException ex) {
System.out.println("消费者读中断");
}
System.out.println("消费对象" + obj);
}
@Override
public void run() {
try {
while (true) {
Object obj = queue.take();
System.out.println("消费者资源队列大小 " + queue.size());
take(obj);
}
} catch (InterruptedException ex) {
System.out.println("消费者中断");
}
}
}
执行者:
public class Main {
public static void main(String args[]) throws InterruptedException {
int numProducers = 4;
int numConsumers = 3;
BlockingQueue<Object> myQueue = new LinkedBlockingQueue<Object>(5);
for (int i = 0; i < numProducers; i++) {
new Thread(new Producer(myQueue)).start();
}
for (int i = 0; i < numConsumers; i++) {
new Thread(new Consumer(myQueue)).start();
}
Thread.sleep(1000);
System.exit(0);
}
}
运行结果如下:
生产者资源队列大小= 1
生产者资源队列大小= 1
消费者 资源 队列大小 1
生产者资源队列大小= 1
消费者 资源 队列大小 1
消费者 资源 队列大小 1
生产者资源队列大小= 1
生产者资源队列大小= 3
消费对象 java.lang.Object@1e1aa52b
生产者资源队列大小= 2
生产者资源队列大小= 5
消费对象 java.lang.Object@6e740a76
消费对象 java.lang.Object@697853f6
......
消费对象 java.lang.Object@41a10cbc
消费对象 java.lang.Object@4963c8d1
消费者 资源 队列大小 5
生产者资源队列大小= 5
生产者资源队列大小= 5
消费者 资源 队列大小 4
消费对象 java.lang.Object@3e49c35d
消费者 资源 队列大小 4
从输出结果中,我们可以发现队列大小永远不会超过5,消费者线程消费了生产者生产的资源。