深入了解ArrayBlockingQueue 阻塞队列
| 阿里云国内75折 回扣 微信号：monov8 |
| 阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6 |
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. 结论
大体的分析就到这里了其实代码还是比较简单的。如果大家有什么疑问及时在评论区留言哦。