java并发之ArrayBlockingQueue详细介绍

java并发之ArrayBlockingQueue详细介绍

java并发之ArrayBlockingQueue详细介绍

ArrayBlockingQueue是常用的线程集合，在线程池中也常常被当做任务队列来使用。使用频率特别高。他是维护的是一个循环队列（基于数组实现），循环结构在数据结构中比较常见，但是在源码实现中还是比较少见的。

线程安全的实现

线程安全队列，基本是离不开锁的。ArrayBlockingQueue使用的是ReentrantLock，配合两种Condition，实现了集合的线程安全操作。这里稍微说一个好习惯，下面是成员变量的声明。

private static final long serialVersionUID = -817911632652898426L;

final Object[] items;

int takeIndex;

int putIndex;

int count;

final ReentrantLock lock;

private final Condition notEmpty;

private final Condition notFull;

transient Itrs itrs = null;

赋值的操作基本都是在构造函数里做的。这样有个好处，代码执行可控。成员变量的初始化也是会合并在构造方法里执行的，但是在执行顺序上需要好好斟酌，如果写在构造方法里初始化，则没有相关问题。

阻塞队列的常用场所就是生产者消费者。一般都是生产者放入，消费者从头取数据。下面重点说这两个操作。

这两个操作都是依靠锁来保证线程安全的。

生产操作

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

}

}

put等放入操作，首先是获取锁，如果发现数据满了，就通过notFull的condition，来阻塞线程。这里的条件判定一定是用while而不是if，多线程情况下，可以被唤醒后发现又满了。

private void enqueue(E x) {

final Object[] items = this.items;

items[putIndex] = x;

if (++putIndex == items.length)

putIndex = 0;

count++;

notEmpty.signal();

}

这个是入队列的操作。首先获取维护的数组。putindex就是放入操作的标志。这个操作会一直加。达到预定的长度后就变成0从头开始计数。这样插入的操作就是一个循环的操作了，count就是用来做计数的，作为能否插入数据的一个标准，插入数据后就通过notEmpty的condition发出一个信号唤醒消费线程。

消费操作

public E take() throws InterruptedException {

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

while (count == 0)

notEmpty.await();

return dequeue();

} finally {

lock.unlock();

}

}

消费的方法也是这样。先获取锁，然后进行条件判断，如果没有数据，则阻塞线程。注意点和put一样。

private E dequeue() {

final Object[] items = this.items;

@SuppressWarnings("unchecked")

E x = (E) items[takeIndex];

items[takeIndex] = null;

if (++takeIndex == items.length)

takeIndex = 0;

count--;

if (itrs != null)

itrs.elementDequeued();

notFull.signal();

return x;

}

取数据的时候，也依靠takeIndex，这是一个标志，这个数值也会一直增加，表示取的第一个数据的位置。如果这个标志走到最后，然后变成0，从头再来。这样保证取出的数据都是fifo的顺序。删除的时候如果发现迭代中，则会修改迭代器的遍历。然后通过notFull的condition来唤醒生产线程。

移除操作

public boolean remove(Object o) {

if (o == null) return false;

final Object[] items = thhttp://is.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;

} finally {

lock.unlock();

}

}

对于remove操作就比较麻烦了，首先获取锁之后，把两个标志位本地化，然后找到要删除的元素的位置。调用removeAt，这里删除需要对标志位做改变。

void removeAt(final int removeIndex) {

final Object[] items = this.items;

if (removeIndex == takeIndex) {

items[takeIndex] = null;

if (++takeIndex == items.length)

takeIndex = 0;

count--;

if (itrs != null)

itrs.elementDequeued();

} else {

final int putIndex = this.putIndex;

for (int i = removeIndex;;) {

int nextwoBMSdCuIy = 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();

}

如果删除的元素是位置和takeindex一样。那就可以直接删除，然后让删除标志位向后移动。如果不是，则从删除的位置开始，进行后面向前面的数据覆盖的操作。直到遇到putindex的前一个位置。然后把那个位置的数据设置为null。并且把putindex的位置往前移动一格，正在迭代的时候要删除数据并且唤醒生产线程。

感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。