ArrayBlockingQueue

ArrayBlockingQueue 是 Java 中的一个有界阻塞队列（bounded blocking queue），它基于数组实现，是线程安全的。它是 java.util.concurrent 包中的一个常用并发工具类，适用于多线程环境下生产者-消费者模型等场景。

一、基本概念

类定义：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable

• 有界性：队列容量固定不可变（构造时指定）。

• 先进先出（FIFO）：元素按插入顺序处理。

• 线程安全：内部使用 ReentrantLock 控制并发访问。

• 阻塞操作：当队列满或空时，某些方法会阻塞等待直到条件满足。

二、核心原理与实现

• 数据结构
  ArrayBlockingQueue 基于固定大小的循环数组实现，通过 takeIndex 和 putIndex 分别标记队列头和尾的位置，支持 FIFO（先进先出）原则35。队列初始化时需指定容量，后续无法动态调整。
• 线程安全机制
  内部通过 ReentrantLock 和两个 Condition（notEmpty、notFull）实现并发控制：
• 生产者阻塞：当队列满时，put() 操作会通过 notFull.await() 阻塞线程，直到有空间；
• 消费者唤醒：成功入队后调用 notEmpty.signal() 唤醒等待的消费者线程。
• 默认使用非公平锁（性能更高），可通过构造函数选择公平锁策略。

三、构造函数

ArrayBlockingQueue(int capacity) 
ArrayBlockingQueue(int capacity, boolean fair)
ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)

• capacity：队列最大容量。

• fair：是否为公平锁（默认为非公平）。如果为 true，则等待时间最长的线程优先获取锁。

• c：初始化集合数据。

四、核心方法说明

五、示例代码

示例 1：基本用法

import java.util.concurrent.ArrayBlockingQueue;

public class ABQDemo {
    public static void main(String[] args) {
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(3);

        try {
            queue.put("A");
            queue.put("B");
            queue.put("C");

            System.out.println(queue.take()); // A
            System.out.println(queue.take()); // B
            System.out.println(queue.take()); // C
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

示例 2：生产者-消费者模式

import java.util.concurrent.ArrayBlockingQueue;

class Producer implements Runnable {
    private ArrayBlockingQueue<Integer> queue;

    public Producer(ArrayBlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                queue.put(i);
                System.out.println("Produced: " + i);
                Thread.sleep(500);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

class Consumer implements Runnable {
    private ArrayBlockingQueue<Integer> queue;

    public Consumer(ArrayBlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                Integer value = queue.take();
                System.out.println("Consumed: " + value);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

public class ProducerConsumerExample {
    public static void main(String[] args) {
        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(3);

        Thread producer = new Thread(new Producer(queue));
        Thread consumer = new Thread(new Consumer(queue));

        producer.start();
        consumer.start();
    }
}

执行结果

Consumed: 0
Produced: 0
Produced: 1
Consumed: 1
Consumed: 2
Produced: 2
Consumed: 3
Produced: 3
Produced: 4
Consumed: 4

六、注意事项

1. 有界限制：一旦创建，容量不可更改。

1. 公平性设置：在高并发下建议开启公平锁（fair = true），避免线程饥饿。
2. 阻塞行为理解清楚：

• put() 和 take() 是阻塞式的，适合用于协调生产者和消费者节奏。

3.异常处理：

• 所有涉及 InterruptedException 的方法必须捕获并恢复中断状态（如 Thread.currentThread().interrupt()）。

七、适用场景

• 任务调度系统：比如多个工作线程从队列中取任务执行。

• 生产者-消费者模型：解耦生产和消费逻辑。

• 限流控制：通过固定大小队列控制资源访问速率。

八、与 LinkedBlockingQueue 的区别

总结

ArrayBlockingQueue 是一个非常适合用于有界缓冲区、任务队列等场景的线程安全队列，尤其适合需要严格控制队列长度的场合。合理使用其阻塞特性可以简化并发编程逻辑。