Java 并发编程 ArrayBlockingQueue 的实现

ArrayBlockingQueue 简介

java.util.concurrent.ArrayBlockingQueue<E> 是 Java 并发编程中的一个阻塞队列，它实现了 BlockingQueue<E> 接口，具有线程安全、高性能、阻塞等特点，由数组实现。

下面我们将介绍 ArrayBlockingQueue 的实现细节和使用方法。

实现原理

ArrayBlockingQueue 的实现是基于数组的，其主要使用以下两个指针来维护队列：

• takeIndex：执行 take() 方法所在位置的索引
• putIndex：执行 put() 方法所在位置的索引

在队列满或空时，使用 ReentrantLock 的条件变量 notFull 和 notEmpty 来阻塞/唤醒线程。

ArrayBlockingQueue 的构造方法

ArrayBlockingQueue 的构造方法如下：

public ArrayBlockingQueue(int capacity, boolean fair)

其中，capacity 为队列容量大小，fair 表示队列是否公平。如果设置 true，则等待时间最长的线程将优先获取队列的访问权。否则线程将随机竞争队列访问权。

ArrayBlockingQueue 的主要方法

ArrayBlockingQueue 实现了 BlockingQueue 接口的一些方法，主要是以下五个常用方法：

1. put(E e)

将元素 e 插入队列的尾部。如果队列已满，则阻塞调用线程，直到队列有空闲位置。

ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
queue.put("element");

2. take()

返回队列头部元素，如果队列为空，则阻塞调用线程，直到队列非空。

ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
String element = queue.take();

3. offer(E e)

将元素 e 插入队列的尾部。如果队列已满，则返回 false。

ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
boolean isSuccess = queue.offer("element");

4. poll()

返回队列头部元素。如果队列为空，则返回 null。

ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
String element = queue.poll();

5. size()

返回队列中元素的个数。

ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
int size = queue.size();

注意事项

1. 当队列已满时，如果调用 put() 方法，调用线程会阻塞，等待队列中有空闲位置。而如果调用 offer() 方法，则直接返回 false。
2. 当队列为空时，如果调用 take() 方法，调用线程会阻塞，等待队列中有元素。而如果调用 poll() 方法，则直接返回 null。

示例说明

示例一

以下示例中，开启两个线程分别向队列中插入元素和从队列中获取元素。因为 ArrayBlockingQueue 是线程安全的，所以这两个线程可以同时访问队列，而且不会出现竞态条件。

public class QueueDemo {
    public static void main(String[] args) throws InterruptedException {
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
        Thread t1 = new Thread(() -> {
            try {
                queue.put("1");
                queue.put("2");
                queue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread t2 = new Thread(() -> {
            try {
                String element1 = queue.take();
                String element2 = queue.take();
                String element3 = queue.take();
                System.out.println(element1);
                System.out.println(element2);
                System.out.println(element3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        t1.start();
        t2.start();
        t1.join();
        t2.join();
    }
}

示例二

以下示例中，开启两个线程分别向队列中插入元素和从队列中获取元素。由于队列是满的，线程1会阻塞并等待队列中有空位置，而线程2会一直阻塞等待队列中有元素。

public class QueueDemo {
    public static void main(String[] args) throws InterruptedException {
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(2);
        Thread t1 = new Thread(() -> {
            try {
                queue.put("1");
                queue.put("2");
                queue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread t2 = new Thread(() -> {
            try {
                String element1 = queue.take();
                String element2 = queue.take();
                String element3 = queue.take();
                System.out.println(element1);
                System.out.println(element2);
                System.out.println(element3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        t1.start();
        t2.start();
        t1.join();
        t2.join();
    }
}