BlockQueue中ArrayBlockingQueue和LinkedBlockingQueue比较

2023-12-10 23:01:57

LinkedBlockingQueue是BlockingQueue的一种使用Link List的实现，它对头和尾（取和添加操作）采用两把不同的锁，相对于ArrayBlockingQueue提高了吞吐量。它也是一种阻塞型的容器，适合于实现“消费者生产者”模式。

ArrayBlockingQueue是对BlockingQueue的一个数组实现，它使用一把全局的锁并行对queue的读写操作，同时使用两个Condition阻塞容量为空时的取操作和容量满时的写操作。

正因为LinkedBlockingQueue使用两个独立的锁控制数据同步，所以可以使存取两种操作并行执行，从而提高并发效率。而ArrayBlockingQueue使用一把锁，造成在存取两种操作争抢一把锁，而使得性能相对低下。LinkedBlockingQueue可以不设置队列容量，默认为Integer.MAX_VALUE.其容易造成内存溢出，一般要设置其值。

LinkedBlockingQueue底层的定义如下：

Java代码

BlockQueue中ArrayBlockingQueue和LinkedBlockingQueue比较

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
static class Node<E> {
/** The item, volatile to ensure barrier separating write and read */
volatile E item;
Node<E> next;
Node(E x) { item = x; }
}
// 支持原子操作
private final AtomicInteger count = new AtomicInteger(0);
// 链表的头和尾
private transient Node<E> head;
private transient Node<E> last;
// 针对取和添加操作的两把锁及其上的条件
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();
...
}

LinkedBlockingQueue的添加操作：

Java代码

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private void insert(E x) {
last = last.next = new Node<E>(x);
}
/**
* signal方法在被调用时，当前线程必须拥有该condition相关的锁!
* Signal a waiting take. Called only from put/offer (which do not
* otherwise ordinarily lock takeLock.)
*/
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
public void put(E o) throws InterruptedException {
if (o == null) throw new NullPointerException();
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
// 使用putLock
putLock.lockInterruptibly();
try {
try {
// 当容量已满时，等待notFull条件
while (count.get() == capacity)
notFull.await();
} catch (InterruptedException ie) {
notFull.signal(); // propagate to a non-interrupted thread
throw ie;
}
insert(o);
// 取出当前值，并将原数据增加1
c = count.getAndIncrement();
// 容量不满，再次激活notFull上等待的put线程
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
// 必须先释放putLock再在notEmpty上signal，否则会造成死锁
if (c == 0)
signalNotEmpty();
}
...
}

LinkedBlockingQueue的取操作：

Java代码

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private E extract() {
Node<E> first = head.next;
head = first;
E x = first.item;
first.item = null;
return x;
}
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
// 使用takeLock
takeLock.lockInterruptibly();
try {
try {
// 若容量为空，等待notEmpty
while (count.get() == 0)
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to a non-interrupted thread
throw ie;
}
x = extract();
c = count.getAndDecrement();
// 再次激活notEmpty
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
// take执行之前容量已满，则激活notFull
if (c == capacity)
signalNotFull();
return x;
}
...
}

ArrayBlockingQueue底层定义如下：

Java代码

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
// 使用循环数组来实现queue，初始时takeIndex和putIndex均为0
private final E[] items;
private transient int takeIndex;
private transient int putIndex;
private int count;
// 用于并发的锁和条件
private final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
/**
* 循环数组
* Circularly increment i.
*/
final int inc(int i) {
return (++i == items.length)? 0 : i;
}
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = (E[]) new Object[capacity];
// 分配锁及该锁上的condition
lock = new ReentrantLock(fair);
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
...
}

ArrayBlockingQueue的取操作：

Java代码

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private E extract() {
final E[] items = this.items;
E x = items[takeIndex];
items[takeIndex] = null;
takeIndex = inc(takeIndex);
--count;
// 激发notFull条件
notFull.signal();
return x;
}
/**
* condition的await的语义如下：
*　与condition相关的锁以原子方式释放，并禁用该线程
*　方法返回时，线程必须获得与该condition相关的锁
*/
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
// 等待notEmpty的条件
while (count == 0)
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to non-interrupted thread
throw ie;
}
E x = extract();
return x;
} finally {
lock.unlock();
}
}
...
}

ArrayBlockingQueue的写操作：

Java代码

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private void insert(E x) {
items[putIndex] = x;
putIndex = inc(putIndex);
++count;
notEmpty.signal();
}
public void put(E o) throws InterruptedException {
if (o == null) throw new NullPointerException();
final E[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
// 等待notFull条件
while (count == items.length)
notFull.await();
} catch (InterruptedException ie) {
notFull.signal(); // propagate to non-interrupted thread
throw ie;
}
insert(o);
} finally {
lock.unlock();
}
}
...
}

注意：ArrayBlockingQueue在读写操作上都需要锁住整个容器，因此吞吐量与一般的实现是相似的，适合于实现“生产者消费者”模式。

通过保证在临界区上多个线程的相互排斥，线程间可以完全避免竞争状态的发生，但是有时候还是需要线程之间的相互协作。使用条件（Condition）便于线程间通信。一个线程可以指定在某种条件下该做什么。标间是通过调用Lock对象的newCoditionn（）方法来实现线程之间的相互通信的。

一旦创建一个条件，就可使用await（）、signal（）、signalAll（）方法来实现线程间通信。await（）方法可以让当前线程都处于等待状态，知道条件放生。signal（）方法唤醒一个等待的线程，而signalAll（）方法唤醒所有等待线程。

注：

Lock接口的线程请求锁的几个方法：

lock(), 拿不到lock就不罢休，不然线程就一直block。比较无赖的做法。
tryLock()，马上返回，拿到lock就返回true，不然返回false。比较潇洒的做法。
带时间限制的tryLock()，拿不到lock，就等一段时间，超时返回false。比较聪明的做法。

下面的lockInterruptibly()就稍微难理解一些。

先说说线程的打扰机制，每个线程都有一个打扰标志。这里分两种情况，
1. 线程在sleep或wait,join，此时如果别的进程调用此进程的 interrupt（）方法，此线程会被唤醒并被要求处理InterruptedException；(thread在做IO操作时也可能有类似行为，见java thread api)
2. 此线程在运行中，则不会收到提醒。但是此线程的 “打扰标志”会被设置，可以通过isInterrupted()查看并作出处理。

lockInterruptibly()和上面的第一种情况是一样的，线程在请求lock并被阻塞时，如果被interrupt，则“此线程会被唤醒并被要求处理InterruptedException”。

引自：http://zhuhui-zj.iteye.com/blog/784193

BlockQueue中ArrayBlockingQueue和LinkedBlockingQueue比较,布布扣,bubuko.com

BlockQueue中ArrayBlockingQueue和LinkedBlockingQueue比较

码农公寓

Lock接口的 线程请求锁的 几个方法：

下面的lockInterruptibly()就稍微难理解一些。

相关文章

Lock接口的线程请求锁的几个方法：