当前位置：首页 > article >正文

java 常用并发队列- ArrayBlockingQueue

article 2024/11/14 2:13:28

1. `ArrayBlockingQueue` 概述

ArrayBlockingQueue 是一个线程安全的、基于数组的有界阻塞队列。它在初始化时需要指定队列的容量，一旦创建，容量就不能改变。它的实现采用了锁和 条件变量 来控制并发访问，并提供了线程间的安全通信机制。

1.1 特性

有界队列：ArrayBlockingQueue 是一个固定大小的队列。队列的容量在创建时设定，不能动态调整。
阻塞操作：put() 和 take() 方法在插入或删除操作受限时（如队列已满或为空）会阻塞调用线程，直到可以继续操作。
线程安全：使用内部锁和条件变量（Condition）来确保线程安全。所有对队列的操作（如插入、删除）都需要获取锁，确保操作的原子性。
FIFO 顺序：队列中的元素按照先进先出（FIFO）的顺序进行排序。

1.2 使用场景

ArrayBlockingQueue 非常适合以下场景：

生产者-消费者模型：在多线程环境下，生产者线程不断将元素插入队列，消费者线程从队列中取出元素进行处理。ArrayBlockingQueue 在生产者和消费者之间充当缓冲区，防止生产者生产过快或消费者消费过快。
任务调度：在线程池或任务调度系统中，用于管理任务队列，确保任务按顺序处理。
资源管理：在资源受限的环境中，可以使用 ArrayBlockingQueue 来管理资源的分配和释放。

2. `ArrayBlockingQueue` 的内部实现

ArrayBlockingQueue 的内部实现基于数组结构，通过 显式锁（ReentrantLock） 和 条件变量（Condition） 来控制并发访问和阻塞操作。以下是其关键的实现细节：

2.1 底层数据结构

ArrayBlockingQueue 使用一个数组 Object[] items 来存储队列元素，同时维护了几个指针来标识队列的状态：

takeIndex：指向下一个要被取出的元素的位置。
putIndex：指向下一个要插入元素的位置。
count：当前队列中的元素数量。

此外，ArrayBlockingQueue 使用了一个 ReentrantLock 对象来实现线程同步，并通过两个条件变量 notEmpty 和 notFull 来控制队列的空满状态：

final Object[] items;
int takeIndex;
int putIndex;
int count;
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;

2.2 插入操作 (`put()` 和 `offer()`)

插入操作向队列尾部添加元素。如果队列已满，put() 方法会阻塞，直到有空间可以插入新元素；offer() 方法则是非阻塞的，如果队列已满，直接返回 false。

以下是 put() 方法的实现：

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();  // 阻塞等待，直到有空间可插入
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

lock.lockInterruptibly()：获取锁并且可以响应中断。
notFull.await()：如果队列已满，当前线程进入等待状态，直到有空间可插入新元素。
enqueue(e)：将元素插入到队列尾部，并更新 putIndex 和 count。

2.3 删除操作 (`take()` 和 `poll()`)

删除操作从队列头部移除元素。如果队列为空，take() 方法会阻塞，直到有可用元素；poll() 方法是非阻塞的，如果队列为空，直接返回 null。

以下是 take() 方法的实现：

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();  // 阻塞等待，直到有可取元素
        return dequeue();
    } finally {
        lock.unlock();
    }
}

notEmpty.await()：如果队列为空，当前线程进入等待状态，直到有元素可以取出。
dequeue()：移除队列头部的元素，并更新 takeIndex 和 count。

2.4 队列中的条件变量（Condition Variables）

ArrayBlockingQueue 使用两个条件变量 notEmpty 和 notFull 来协调生产者和消费者线程：

notFull：当队列已满时，生产者线程调用 put() 方法会阻塞在 notFull.await() 处。只有当队列有空间时，调用 notFull.signal() 唤醒等待线程。
notEmpty：当队列为空时，消费者线程调用 take() 方法会阻塞在 notEmpty.await() 处。只有当队列中有元素时，调用 notEmpty.signal() 唤醒等待线程。

3. `ArrayBlockingQueue` 的使用示例

以下是一个典型的 ArrayBlockingQueue 使用示例，演示了生产者-消费者模型的应用：

import java.util.concurrent.ArrayBlockingQueue;

public class ProducerConsumerExample {
    public static void main(String[] args) {
        // 创建一个容量为 5 的 ArrayBlockingQueue
        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);

        // 生产者线程
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    System.out.println("Producing: " + i);
                    queue.put(i);  // 阻塞插入
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 消费者线程
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    Integer value = queue.take();  // 阻塞移除
                    System.out.println("Consuming: " + value);
                    Thread.sleep(150);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 启动线程
        producer.start();
        consumer.start();
    }
}

在这个示例中，生产者线程将整数插入到队列中，消费者线程从队列中取出整数。ArrayBlockingQueue 确保了生产者和消费者线程之间的安全通信和同步。

4. `ArrayBlockingQueue` 的优缺点

4.1 优点

线程安全：通过显式锁和条件变量来确保并发操作的线程安全性。
高效的阻塞机制：支持阻塞的插入和删除操作，适用于生产者-消费者模型。
固定容量控制：通过指定容量来防止资源耗尽，适合需要限制队列大小的场景。

4.2 缺点

固定容量：容量在初始化时设定，不能动态调整，不适合需要动态扩展的场景。
锁竞争：在高并发环境下，多个线程竞争锁可能会导致性能下降。
非公平性：默认情况下，ArrayBlockingQueue 采用非公平锁策略，可能会导致线程饥饿问题。

5. 与其他阻塞队列的对比

LinkedBlockingQueue：基于链表的阻塞队列，可以选择无界或有界，适合需要较大队列容量或不确定容量的场景。相比之下，ArrayBlockingQueue 使用固定容量数组，内存利用率更高，适合容量已知的场景。
PriorityBlockingQueue：基于优先级的无界阻塞队列，适合需要按优先级处理任务的场景。ArrayBlockingQueue 则是基于 FIFO 顺序的固定容量队列。

6. 总结

ArrayBlockingQueue 是 Java 并发编程中一种常用的有界阻塞队列，适合生产者-消费者模型及需要在多个线程之间安全地共享数据的场景。通过显式锁和条件变量，ArrayBlockingQueue 提供了简单而有效的线程间通信机制。尽管它有容量限制和可能的锁竞争问题，但在需要固定容量控制和阻塞操作的场景中，ArrayBlockingQueue 是一个非常实用的选择。
ArrayBlockingQueue 是 Java 中常用的并发队列之一，属于阻塞队列的范畴。它是一个 基于数组的有界阻塞队列，实现了 BlockingQueue 接口。ArrayBlockingQueue 在多线程环境中能够有效地管理线程之间的数据传递与任务协调，广泛应用于生产者-消费者模型和各种需要在多个线程之间进行安全数据共享的场景中。

查看全文

http://www.kler.cn/a/289640.html