您的位置:首页 > 编程语言 > Java开发

JAVA并发(6)-并发队列ArrayBlockingQueue

2021-06-09 20:34 120 查看 https://www.cnblogs.com/ukyu/p

本文讲ArrayBlockingQueue

1. 介绍

一个基于数组的有界阻塞队列,FIFO顺序。支持等待消费者和生产者线程的可选公平策略(默认是非公平的)。公平的话通常会降低吞吐量,但是可以减少可变性并避免之前被阻塞的线程饥饿。

1.1 类结构

  • ArrayBlockingQueue继承关系

  • ArrayBlockingQueue类图

构造器

// 默认是非公平的
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}

public ArrayBlockingQueue(int capacity, boolean fair) {
...
}

public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
...
}

比较重要的几个参数

// 储存元素的数组
final Object[] items;

/** items index for next take, poll, peek or remove */
// 与putIndex相互配合可以将数组变成一个可循环利用的数组,不需要扩容,后面会讲到
// 每次出队的索引
int takeIndex;

/** items index for next put, offer, or add */
// 每次入队的索引
int putIndex;

/** Number of elements in the queue */
int count;

/**
* Shared state for currently active iterators, or null if there
* are known not to be any.  Allows queue operations to update
* iterator state.
*/
// 迭代的时候会用到,在后面详讲
transient Itrs itrs = null;

保证线程安全的措施

/** Main lock guarding all access */
final ReentrantLock lock;

/** Condition for waiting takes */
private final Condition notEmpty;

/** Condition for waiting puts */
private final Condition notFull;

我们可以看到ArrayBlockingQueue使用的是单锁控制线程安全,而LinkedBlockingQueue双锁控制的, 后者的细粒度更小。

2. 源码剖析

ArrayBlockingQueue也是继承至BlockingQueue(可以去看看上面提到的那篇博客有提到BlockingQueue),它对于不同的方法不能立即满足要求的,作出的回应是不一样的。

我们分别介绍下面的方法的具体实现

  • offer(E e)
  • offer(E e, long timeout, TimeUnit unit)
  • put(E e)
  • poll()
  • remove(Object o)

2.1 offer(E e) & poll()

插入成功就返回true;若队列满了就直接返回false,不会阻塞自己

public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} finally {
lock.unlock();
}
}

上面的代码比较简单,我们来看看入队的具体操作

private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;

// 为什么putIndex+1 等于数组长度时会变成0
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}

为了解答上面注释中的问题,我们先看看poll()的实现

public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;

// takeIndex + 1等于了数组的长度也会将值置为0
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();
return x;
}

结合上面的入队、出队源码,我们来分析一下:

  • 单线程下,首先执行
ArrayBlockingQueue<String> array = new ArrayBlockingQueue<>(3);
array.offer("A");
array.offer("B");
array.offer("C");

此时队列的状态

  • 再执行
array.poll();
array.offer("D");

最后队列的状态

大家可能会有点疑问,上面的队列不是输出是"D B C", 咋回事? 肯定不是啦,我们看看类重写的toString就明白了。

public String toString() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
int k = count;
if (k == 0)
return "[]";

final Object[] items = this.items;
StringBuilder sb = new StringBuilder();
sb.append('[');

// 主要代码
for (int i = takeIndex; ; ) {
Object e = items[i];
sb.append(e == this ? "(this Collection)" : e);
if (--k == 0)
return sb.append(']').toString();
sb.append(',').append(' ');
if (++i == items.length)
i = 0;
}
} finally {
lock.unlock();
}
}

思考一下,就会明白了。
通过上面的分析,我们看出了数组就像一个循环数组一样,每个地址都被重复使用。我们也知道了基于数组的队列如何实现的

offer(E e, long timeout, TimeUnit unit)put(E e)实现都比较简单,大家看看源码即可。

2.2 remove(Object o)

若o存在则移除,返回true;反之。这个操作会改变队列的结构,但是该方法一般很少使用

public boolean remove(Object o) {
if (o == null) return false;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count > 0) {
final int putIndex = this.putIndex;
int i = takeIndex;
do {
if (o.equals(items[i])) {
// 主要删除逻辑
removeAt(i);
return true;
}
if (++i == items.length)
i = 0;
} while (i != putIndex);
}
return false;

ad0
} finally {
lock.unlock();
}
}
void removeAt(final int removeIndex) {
// assert lock.getHoldCount() == 1;
// assert items[removeIndex] != null;
// assert removeIndex >= 0 && removeIndex < items.length;
final Object[] items = this.items;
if (removeIndex == takeIndex) {
// removing front item; just advance
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
} else {
// an "interior" remove

// slide over all others up through putIndex.
// 此时removeIndex != takeIndex
// 为啥要执行下面的代码,大家可以按照上面图片的最后状态,
// 按照下面代码走一下,就明白了.主要是设置putIndex
final int putIndex = this.putIndex;
for (int i = removeIndex;;) {
int next = i + 1;
if (next == items.length)
next = 0;
if (next != putIndex) {
items[i] = items[next];
i = next;
} else {
items[i] = null;
this.putIndex = i;
break;
}
}
count--;
if (itrs != null)
itrs.removedAt(removeIndex);
}
notFull.signal();
}

2.3 解释解释Itrs

// 当前活动迭代器的共享状态; 允许队列操作更新迭代器的状态;
transient Itrs itrs = null;

这个变量可以理解成,在一个线程使用迭代器时,其他的线程可以对队列进行更新操作的一个保障。
源码注释中对Itrs的描述,迭代器和它们的队列之间共享数据,允许在删除元素时修改队列以更新迭代器。 我们可以看到对队列进行了删除操作时,队列都会执行下面的语句

if (itrs != null)
itrs.removedAt(removeIndex);

初始化该值是在使用迭代器时

public Iterator<E> iterator() {
return new Itr();
}

...

Itr() {
// assert lock.getHoldCount() == 0;
lastRet = NONE;
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
...
itrs = new Itrs(this);

ad8
...
}
} finally {
lock.unlock();
}
}

3. 总结

ArrayBlockingQueue的实现整体不难,使用ReetrantLock保证了线程安全,putIndextakeIndex分别维护入队与出队的位置,一起构成一个循环数组

内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: