java并发编程(二十四)----(JUC集合)ArrayBlockingQueue和LinkedBlockingQueue介绍

时间:2024-09-30 08:06:37

这一节我们来了解阻塞队列(BlockingQueue),BlockingQueue接口定义了一种阻塞的FIFO queue,每一个BlockingQueue都有一个容量,当容量满时往BlockingQueue中添加数据时会造成阻塞,当容量为空时取元素操作会阻塞。首先我们来看ArrayBlockingQueue和LinkedBlockingQueue.

ArrayBlockingQueue

ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此队列按照先进先出(FIFO)的原则对元素进行排序。默认情况下不保证访问者公平的访问队列,所谓公平访问队列是指阻塞的所有生产者线程或消费者线程,当队列可用时,可以按照阻塞的先后顺序访问队列,即先阻塞的生产者线程,可以先往队列里插入元素,先阻塞的消费者线程,可以先从队列里获取元素。通常情况下为了保证公平性会降低吞吐量。

我们看他的构造函数实现:

//默认是非公平的,初始指定队列容量
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
} //该构造方法可以设置队列的公平性。当然如果为公平的,则对性能会产生影响
//访问者的公平性是使用可重入锁实现的
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity];
lock = new ReentrantLock(fair);
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}

使用很简单我们直接看一个实例:

public class ProducerConsumerTest {
public static void main(String[] args) { final BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(3); ExecutorService service = Executors.newFixedThreadPool(10); for(int i = 0;i<4;i++){
service.execute(new ProducerAndConsumer(blockingQueue));
}
}
} class ProducerAndConsumer implements Runnable{
private boolean flag = false; private Integer j = 1; private Lock lock = new ReentrantLock(); Condition pro_con = lock.newCondition(); Condition con_con = lock.newCondition(); private BlockingQueue<Integer> blockingQueue; public ProducerAndConsumer(BlockingQueue<Integer> blockingQueue){
this.blockingQueue= blockingQueue;
} //生产
public void put(){
try {
lock.lock();
while(flag)
pro_con.await();
System.out.println("正在准备放入数据。。。");
Thread.sleep(new Random().nextInt(10)*100);
Integer value = new Random().nextInt(30);
blockingQueue.put(value);
System.out.println(Thread.currentThread().getName()+" 放入的数据 "+value);
flag = true;
con_con.signal();
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
} public void get(){
try {
lock.lock();
while(!flag)
con_con.await();
System.out.println("正在准备取数据。。。");
Thread.sleep(new Random().nextInt(10)*1000);
System.out.println(Thread.currentThread().getName()+" 取到的数据为"+blockingQueue.take());
flag = false;
pro_con.signal();
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
} @Override
public void run() { while(true){
if(j==1){
put();
}
else{
get();
}
j=(j+1)%2;
}
}
}

输出为:

正在准备放入数据。。。
正在准备放入数据。。。
正在准备放入数据。。。
正在准备放入数据。。。
pool-1-thread-2 放入的数据 13
正在准备取数据。。。
pool-1-thread-3 放入的数据 4
正在准备取数据。。。
pool-1-thread-3 取到的数据为13
正在准备放入数据。。。
pool-1-thread-1 放入的数据 11
正在准备取数据。。。
pool-1-thread-4 放入的数据 26
正在准备取数据。。。
pool-1-thread-1 取到的数据为4
正在准备放入数据。。。
pool-1-thread-2 取到的数据为11
正在准备放入数据。。。
pool-1-thread-3 放入的数据 18
正在准备取数据。。。
...
...

LinkedBlockingQueue

LinkedBlockingQueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。

先看一下他的构造函数:

public LinkedBlockingQueue() {
this(Integer.MAX_VALUE); //MAX_VALUE=2147483647
} public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}

我们还是直接开看一个例子:

public class BlockingQueueTest {

    /**
* 定义装苹果的篮子
*/
public static class Basket {
// 篮子,能够容纳3个苹果
// BlockingQueue<String> basket = new ArrayBlockingQueue<String>(3);
BlockingQueue<String> basket = new LinkedBlockingQueue<String>(3); // 生产苹果,放入篮子
public void produce() throws InterruptedException {
// put方法放入一个苹果,若basket满了,等到basket有位置
basket.put("An apple");
} // 消费苹果,从篮子中取走
public String consume() throws InterruptedException {
// get方法取出一个苹果,若basket为空,等到basket有苹果为止
return basket.take();
}
} //  测试方法
public static void testBasket() { // 建立一个装苹果的篮子
final Basket basket = new Basket(); // 定义苹果生产者
class Producer implements Runnable {
public String instance = ""; public Producer(String a) {
instance = a;
} public void run() {
try {
while (true) {
// 生产苹果
System.out.println("生产者准备生产苹果:" + instance);
basket.produce();
System.out.println("! 生产者生产苹果完毕:" + instance);
// 休眠300ms
Thread.sleep(300);
}
} catch (InterruptedException ex) {
}
}
} // 定义苹果消费者
class Consumer implements Runnable {
public String instance = ""; public Consumer(String a) {
instance = a;
} public void run() {
try {
while (true) {
// 消费苹果
System.out.println("消费者准备消费苹果:" + instance);
basket.consume();
System.out.println("! 消费者消费苹果完毕:" + instance);
// 休眠1000ms
Thread.sleep(1000);
}
} catch (InterruptedException ex) {
}
}
} ExecutorService service = Executors.newCachedThreadPool();
Producer producer = new Producer("P1");
Producer producer2 = new Producer("P2");
Consumer consumer = new Consumer("C1");
service.submit(producer);
service.submit(producer2);
service.submit(consumer); // 程序运行3s后,所有任务停止
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
} service.shutdownNow();
} public static void main(String[] args) {
BlockingQueueTest.testBasket();
}
}

输出为:

生产者准备生产苹果:P1
消费者准备消费苹果:C1
! 生产者生产苹果完毕:P1
生产者准备生产苹果:P2
! 消费者消费苹果完毕:C1
! 生产者生产苹果完毕:P2
生产者准备生产苹果:P2
! 生产者生产苹果完毕:P2
生产者准备生产苹果:P1
! 生产者生产苹果完毕:P1
生产者准备生产苹果:P2
生产者准备生产苹果:P1
消费者准备消费苹果:C1
! 消费者消费苹果完毕:C1
! 生产者生产苹果完毕:P2
生产者准备生产苹果:P2
消费者准备消费苹果:C1
! 消费者消费苹果完毕:C1
! 生产者生产苹果完毕:P1
生产者准备生产苹果:P1
消费者准备消费苹果:C1
! 消费者消费苹果完毕:C1
! 生产者生产苹果完毕:P2 Process finished with exit code 0