详解ArrayBlockQueue源码解析

详解ArrayBlockingQueue源码解析

ArrayBlockingQueue是Java集合框架中的阻塞队列，该队列的容量固定不变，而且是有界的。它是线程安全的，任何时刻只有一个线程能够访问队列，当队列已满时插入元素的线程会被阻塞，当队列为空时，获取元素的线程会被阻塞。

基本特性

• 固定容量大小
• 先进先出
• 线程安全
• 阻塞队列

主要方法

ArrayBlockingQueue继承了AbstractQueue、AbstractCollection等接口中定义的方法，主要方法都位于ArrayBlockingQueue内部的private方法中。

插入元素

put(E e)：将指定元素插入队列尾部。如果队列已满，则调用该方法的线程将会被阻塞。

移除元素

take()：获取并移除队列头部元素。如果队列为空，则调用该方法的线程将会被阻塞。

查询队列中元素的个数

size()：返回队列中元素的个数。

源码分析

构造方法

构造方法用于初始化队列，需要传入队列的容量以及是否公平锁，默认为非公平锁。

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();
}

插入元素

put(E e)方法用于将元素插入队列尾部，当队列已满时会导致调用该方法的线程被阻塞。

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly(); // 公平锁和非公平锁都支持中断
    try {
        while (count == items.length)
            notFull.await(); // 如果队列已满，则等待
        insert(e); // 插入元素
    } finally {
        lock.unlock();
    }
}

移除元素

take()方法用于获取并移除队列头部元素，当队列为空时会导致调用该方法的线程被阻塞。

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await(); // 如果队列为空，则等待
        return extract();
    } finally {
        lock.unlock();
    }
}

查询队列中元素的个数

size()方法用于返回队列中元素的个数。

public int size() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return count;
    } finally {
        lock.unlock();
    }
}

示例说明

示例 1：

ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
new Thread(() -> {
    for (int i = 0; i < 100; i++) {
        try {
            queue.put(i);
            System.out.println("生产者生产了：" + i);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}).start();
new Thread(() -> {
    for (int i = 0; i < 100; i++) {
        try {
            Integer num = queue.take();
            System.out.println("消费者消费了：" + num);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}).start();

上述代码创建了一个容量为10的ArrayBlockingQueue，启动生产者线程向队列中生产100个整数，启动消费者线程从队列中消费100个整数。当生产者生产到10个之后，由于队列已满，会导致生产者线程被阻塞，直到队列中的元素被消费到不到10个时，生产者线程才能够继续生产。

示例 2：

ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
new Thread(() -> {
    for (int i = 0; i < 100; i++) {
        try {
            queue.put(i);
            System.out.println("生产者生产了：" + i);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}).start();
new Thread(() -> {
    for (int i = 0; i < 100; i++) {
        try {
            Integer num = queue.take();
            System.out.println("消费者消费了：" + num);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}).start();

上述代码同样创建了一个容量为10的ArrayBlockingQueue，启动生产者线程向队列中生产100个整数，启动消费者线程从队列中消费100个整数。当消费者从队列中取走所有元素后，队列为空，会导致消费者线程被阻塞，直到队列中又生产一个元素后，消费者线程才能继续消费。