1. 前言
开始正式了解阻塞队列之前，我们需要了解什么是
队列。队列有什么作用。其实队列的作用就是解耦，更加确切的说应该是生产者以及消费者 之间的解耦
今天就让我们来看下
ArrayBlockingQueue的实现。虽然通过名称就可以看到，无非是通过数组进行实现但是我们还是要剥离下Java底层代码是如何实现的。
2. 简单示例
package com.lihh.queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
public class T01_Queue_Test01 {
public static void main(String[] args) throws InterruptedException {
// addTest();
// offerTest();
// offerTest01();
putTest();
}
public static void putTest() throws InterruptedException {
ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(2);
arrayBlockingQueue.put("1");
arrayBlockingQueue.put("2");
new Thread(() -> {
try {
Thread.sleep(2000);
arrayBlockingQueue.remove();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}).start();
// 一直处于阻塞状态
arrayBlockingQueue.put("3");
System.out.println(arrayBlockingQueue);
}
public static void offerTest01() throws InterruptedException {
ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(2);
System.out.println(arrayBlockingQueue.offer("1"));
System.out.println(arrayBlockingQueue.offer("2"));
new Thread(() -> {
try {
Thread.sleep(1000);
System.out.println(arrayBlockingQueue.remove());
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}).start();
// 此处表示指定时间范围内 时间是阻塞状态
System.out.println(arrayBlockingQueue.offer("3", 3, TimeUnit.SECONDS));
}
public static void offerTest() {
ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(2);
System.out.println(arrayBlockingQueue.offer("1"));
System.out.println(arrayBlockingQueue.offer("2"));
// 如果队列的长度已满 ，无法添加新的元素，直接返回false
System.out.println(arrayBlockingQueue.offer("3"));
}
public static void addTest() {
ArrayBlockingQueue<String> strings = new ArrayBlockingQueue<>(2);
strings.add("1");
strings.add("2");
// 如果超过队列的长度 直接报错
strings.add("3");
}
}
3. Java层面分析
3.1 constructor 实现
3.1.1 定义
其实通过构造函数重写就可以看到，第一个参数一定是一个int类型的值。 既然底层是基于数组进行实现的。所以数组必须限定一个长度，就是这个所谓的int类型所代表的值。
3.1.2 内容
// 数组阻塞队列的实现方法。
// capacity 表示数组的长度
// fair 表示是公平锁 还是 非公平锁
public ArrayBlockingQueue(int capacity, boolean fair) {
// 如果长度小于等于0的话 直接报异常错误
if (capacity <= 0)
throw new IllegalArgumentException();
// 实例化一个对象数组
this.items = new Object[capacity];
// 实例化一个锁
lock = new ReentrantLock(fair);
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
3.2 生产者实现方法
3.2.1 add 实现方法
通过内部API
offer来添加元素。添加成功后返回true。反之就会报错。(如果队列中元素满了也会添加失败，也会报错误)
public boolean add(E e) {
// 内部直接调用offer方法，如果元素添加成功后返回true。 反之直接就是抛出异常
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
3.2.2 offer 实现方法
此方法相对简单，就往队列中添加元素，如果添加成功就返回true，反之就返回false。
// 添加元素的方法
public boolean offer(E e) {
// 判断是否为空值
checkNotNull(e);
// 获取锁示例
final ReentrantLock lock = this.lock;
// 枷锁
lock.lock();
try {
// 如果添加的元素的个数 == 数组的长度了。 直接返回false
if (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} finally {
// 解锁
lock.unlock();
}
}
private void enqueue(E x) {
// 成员 => 局部
final Object[] items = this.items;
// 给指定位置 设置元素。 变量putIndex 控制添加元素移动的
items[putIndex] = x;
if (++putIndex == items.length)
//
putIndex = 0;
// 元素个数累加的
count++;
// 将Condition中 阻塞的线程唤醒。
notEmpty.signal();
}
3.2.3 有参offer实现方法
跟
offer(int)的用法保持一致。但是唯一不同的是：如果队列中元素满了后，线程会被阻塞一定的时间，如果时间到期后还不能添加到队列中，就直接返回false
// 添加元素 但是会有一定时间的阻塞
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
// 判断是否为空
checkNotNull(e);
// 统一转换时间
long nanos = unit.toNanos(timeout);
// 获取锁
final ReentrantLock lock = this.lock;
// 设置可 中断锁
lock.lockInterruptibly();
try {
// 如果队列中的元素已满的话 while等待中
while (count == items.length) {
// 如果等待时间到了 直接返回false
if (nanos <= 0)
return false;
// 挂起线程，同时释放资源
// awaitNanos会挂起线程，并且返回剩余的阻塞时间
// 恢复执行时，需要重新获取锁资源
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
3.2.3 put 实现方法
也是往队列中添加元素的方法，如果队列中元素满了，会一直处于阻塞状态，直到可以添加到底队列为止。
// 添加元素
public void put(E e) throws InterruptedException {
// 判断是否为空
checkNotNull(e);
// 获取锁实例
final ReentrantLock lock = this.lock;
// 加 可打断锁
lock.lockInterruptibly();
try {
// await方法一直阻塞，直到被唤醒或者中断标记位
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
// 删除锁
lock.unlock();
}
}
3.3 消费者实现方法
3.3.1 remove 实现方法
依据队列的特性进行元素删除，当队列中没有元素了，直接报异常
// 删除元素的方法
public E remove() {
// 内部直接使用poll方法来实现
E x = poll();
// 如果结果不为null的话 直接返回
if (x != null)
return x;
else
// 反之就是报异常
throw new NoSuchElementException();
}
3.3.2 poll 实现方法
删除元素的API，如果队列中没有元素了直接返回null
// 删除元素的方法
public E poll() {
// 获取锁
final ReentrantLock lock = this.lock;
// 进行加锁
lock.lock();
try {
// 如果队列中没有数据了 直接返回null。 反之就是删除数据
return (count == 0) ? null : dequeue();
} finally {
// 解锁
lock.unlock();
}
}
private E dequeue() {
// 保存元素的队列
final Object[] items = this.items;
@SuppressWarnings("unchecked")
// 获取将要剔除的元素
E x = (E) items[takeIndex];
// 将剔除的元素重置为null
items[takeIndex] = null;
// 如果已经删除的是队列的最后一个元素了 将下标设置为第一个元素
if (++takeIndex == items.length)
takeIndex = 0;
// 元素递减
count--;
if (itrs != null)
itrs.elementDequeued();
// signal方法，会唤醒当前Condition中排队的一个Node。
notFull.signal();
return x;
}
3.3.3 有参poll实现方法
删除元素的API，当队列中没有数据的时候，可以阻塞等待指定时间，时间到期后队列中还没有数据的话，直接返回null
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
// 获取阻塞的时间
long nanos = unit.toNanos(timeout);
// 获取锁
final ReentrantLock lock = this.lock;
// 加锁
lock.lockInterruptibly();
try {
// while循环 直到存在数据跳出循环
while (count == 0) {
// 如果时间到期了 直接跳过
if (nanos <= 0)
return null;
// 没数据，挂起消费者线程
nanos = notEmpty.awaitNanos(nanos);
}
return dequeue();
} finally {
lock.unlock();
}
}
3.3.4 take 实现方法
删除元素的API，如果队列中没有元素了会一直挂起线程等待。
public E take() throws InterruptedException {
// 获取锁
final ReentrantLock lock = this.lock;
// 设置加锁
lock.lockInterruptibly();
try {
// 线程挂起，直到队列中出现元素
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
4. 结论
大体的分析就到这里了，其实代码还是比较简单的。如果大家有什么疑问，及时在评论区留言哦。
