Java集合框架ArrayBlockingQueue应用分析

一、ArrayBlockingQueue介绍

ArrayBlockingQueue是Java集合框架中的一种阻塞队列，它是线程安全的有限队列，底层是使用数组实现的，它具有FIFO（先进先出）的特性，支持高效的并发访问。ArrayBlockingQueue是一种固定长度的队列，当队列已满时，会阻塞插入元素的线程，当队列已空时，会阻塞获取元素的线程，保证线程安全。

ArrayBlockingQueue的构造方法如下：

public ArrayBlockingQueue(int capacity, boolean fair);

其中，capacity为队列容量，fair为是否保证公平性，即使用先进先出的方式获取元素。

ArrayBlockingQueue的常用方法：

• put(E e)：将元素插入队尾（如果队列已满则阻塞）
• take()：获取队头元素（如果队列已空则阻塞）
• offer(E e, long timeout, TimeUnit unit)：在指定时间内尝试将元素插入队尾
• poll(long timeout, TimeUnit unit)：在指定时间内尝试获取队头元素
• size()：获取队列元素个数

二、示例说明

1. 生产者和消费者模型

ArrayBlockingQueue通常应用于生产者和消费者模型，其中生产者不断地往队列中添加元素，消费者从队列中获取元素并进行处理。

下面是一个简单的示例代码：

import java.util.concurrent.ArrayBlockingQueue;

public class ProducerConsumerExample {

    public static void main(String[] args) {

        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(10);

        // 生产者线程
        Thread producerThread = new Thread(() -> {
            int count = 0;
            while (true) {
                try {
                    String message = "Message " + count++;
                    queue.put(message);
                    System.out.println("Produced: " + message);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        producerThread.start();

        // 消费者线程
        Thread consumerThread = new Thread(() -> {
            while (true) {
                try {
                    String message = queue.take();
                    System.out.println("Consumed: " + message);
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        consumerThread.start();
    }
}

该示例中，生产者线程往队列中添加了10个元素，然后就一直在队列已满时阻塞等待；消费者线程从队列中获取元素并输出，然后就一直在队列已空时阻塞等待。

2. 多生产者和多消费者模型

ArrayBlockingQueue还可以应用于多生产者和多消费者模型，其中多个生产者不断地往队列中添加元素，多个消费者从队列中获取元素并进行处理。

下面是一个简单的示例代码：

import java.util.concurrent.ArrayBlockingQueue;

public class MultiProducerConsumerExample {

    public static void main(String[] args) {

        // 创建一个容量为10的队列
        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

        // 创建2个生产者线程
        Thread producerThread1 = new Thread(() -> {
            int count = 0;
            while (true) {
                try {
                    queue.put(count);
                    System.out.println(Thread.currentThread().getName() + " produced: " + count++);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "Producer 1");
        producerThread1.start();

        Thread producerThread2 = new Thread(() -> {
            int count = 0;
            while (true) {
                try {
                    queue.put(count);
                    System.out.println(Thread.currentThread().getName() + " produced: " + count++);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "Producer 2");
        producerThread2.start();

        // 创建3个消费者线程
        Thread consumerThread1 = new Thread(() -> {
            while (true) {
                try {
                    int value = queue.take();
                    System.out.println(Thread.currentThread().getName() + " consumed: " + value);
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "Consumer 1");
        consumerThread1.start();

        Thread consumerThread2 = new Thread(() -> {
            while (true) {
                try {
                    int value = queue.take();
                    System.out.println(Thread.currentThread().getName() + " consumed: " + value);
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "Consumer 2");
        consumerThread2.start();

        Thread consumerThread3 = new Thread(() -> {
            while (true) {
                try {
                    int value = queue.take();
                    System.out.println(Thread.currentThread().getName() + " consumed: " + value);
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "Consumer 3");
        consumerThread3.start();
    }
}

该示例中，创建了2个生产者线程和3个消费者线程，它们不断地往队列中添加元素和从队列中获取元素，并进行输出。当队列已满或已空时，线程会自动阻塞等待。

三、总结

ArrayBlockingQueue是一个非常有用的并发集合，通常应用于生产者和消费者模型以及多生产者和多消费者模型等场景下。通过合理地使用ArrayBlockingQueue，我们可以避免线程安全问题，实现高效的并发访问。