集合框架 Queue篇（6）—LinkedBlockingQueue

来源：互联网

Queue

------------
1.ArrayDeque, （数组双端队列）
2.PriorityQueue, （优先级队列）
3.ConcurrentLinkedQueue, （基于链表的并发队列）

4.DelayQueue,                                         （延期阻塞队列）（阻塞队列实现了BlockingQueue接口）
5.ArrayBlockingQueue,           （基于数组的并发阻塞队列）
6.LinkedBlockingQueue,        （基于链表的FIFO阻塞队列）
7.LinkedBlockingDeque, （基于链表的FIFO双端阻塞队列）
8.PriorityBlockingQueue,        （带优先级的无界阻塞队列）
9.SynchronousQueue                       （并发同步阻塞队列）
-----------------------------------------------------
LinkedBlockingQueue

采用对于的next构成链表的方式来存储对象。
由于读只操作队头，而写只操作队尾，这里巧妙地采用了两把锁，
对put和offer采用putLock，对take和poll采用takeLock，避免了读写时相互竞争锁的现象，
因此LinkedBlockingQueue在高并发读写操作都多的情况下，性能会比ArrayBlockingQueue好很多，
在遍历以及删除元素时则要把两把锁都锁住。

public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
/**
* 链表节点node类结构
*/
static class Node<E> {
volatile E item;//volatile使得所有的write happen-befor read，保证了数据的可见性
Node<E> next;
Node(E x) { item = x; }
}

/** 队列容量，默认为Integer.MAX_VALUE*/
private final int capacity;

/** 用原子变量表示当前元素的个数 */
private final AtomicInteger count = new AtomicInteger(0);

/** 表头节点 */
private transient Node<E> head;

/** 表尾节点 */
private transient Node<E> last;

/** 获取元素或删除元素时要加的takeLock锁 */
private final ReentrantLock takeLock = new ReentrantLock();

/** 获取元素 notEmpty条件 */
private final Condition notEmpty = takeLock.newCondition();

/** 插入元素时要加putLock锁 */
private final ReentrantLock putLock = new ReentrantLock();

/** 插入时，要判满 */
private final Condition notFull = putLock.newCondition();

/**
* 唤醒等待的take操作，在put/offer中调用（因为这些操作中不会用到takeLock锁）
*/
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}

/**
* 唤醒等待插入操作，在take/poll中调用.
*/
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}

/**
* 插入到尾部
*/
private void insert(E x) {
last = last.next = new Node<E>(x);
}

/**
* 获取并移除头元素
*/
private E extract() {
Node<E> first = head.next;
head = first;
E x = first.item;
first.item = null;
return x;
}

/**
* 锁住两把锁，在remove，clear等方法中调用
*/
private void fullyLock() {
putLock.lock();
takeLock.lock();
}

/**
* 和fullyLock成对使用
*/
private void fullyUnlock() {
takeLock.unlock();
putLock.unlock();
}

/**
* 默认构造，容量为 Integer.MAX_VALUE
*/
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
/**
*指定容量的构造
*/
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}
/**
* 指定初始化集合的构造
*/
public LinkedBlockingQueue(Collection<? extends E> c) {
this(Integer.MAX_VALUE);
for (E e : c)
add(e);
}

/**
* 通过原子变量，直接获得大小
*/
public int size() {
return count.get();
}

/**
*返回理想情况下（没有内存和资源约束）此队列可接受并且不会被阻塞的附加元素数量。
*/
public int remainingCapacity() {
return capacity - count.get();
}

/**
* 将指定元素插入到此队列的尾部，如有必要，则等待空间变得可用。
*/
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
try {
while (count.get() == capacity)
notFull.await();
} catch (InterruptedException ie) {
notFull.signal(); // propagate to a non-interrupted thread
throw ie;
}
insert(e);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}

/**
* 将指定元素插入到此队列的尾部，如有必要，则等待指定的时间以使空间变得可用。
*/
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {

if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
for (;;) {
if (count.get() < capacity) {
insert(e);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
break;
}
if (nanos <= 0)
return false;
try {
nanos = notFull.awaitNanos(nanos);
} catch (InterruptedException ie) {
notFull.signal(); // propagate to a non-interrupted thread
throw ie;
}
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return true;
}

/**
*将指定元素插入到此队列的尾部（如果立即可行且不会超出此队列的容量），
*在成功时返回 true，如果此队列已满，则返回 false。
*/
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;
int c = -1;
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
if (count.get() < capacity) {
insert(e);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return c >= 0;
}

//获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）。
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
try {
while (count.get() == 0)
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to a non-interrupted thread
throw ie;
}

x = extract();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
//获取并移除此队列的头部，在指定的等待时间前等待可用的元素（如果有必要
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E x = null;
int c = -1;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
for (;;) {
if (count.get() > 0) {
x = extract();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
break;
}
if (nanos <= 0)
return null;
try {
nanos = notEmpty.awaitNanos(nanos);
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to a non-interrupted thread
throw ie;
}
}
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
//获取并移除此队列的头，如果此队列为空，则返回 null。
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
if (count.get() > 0) {
x = extract();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
}
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}

//获取但不移除此队列的头；如果此队列为空，则返回 null。
public E peek() {
if (count.get() == 0)
return null;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
Node<E> first = head.next;
if (first == null)
return null;
else
return first.item;
} finally {
takeLock.unlock();
}
}

/**
* 从此队列移除指定元素的单个实例（如果存在）。
*/
public boolean remove(Object o) {
if (o == null) return false;
boolean removed = false;
fullyLock();
try {
Node<E> trail = head;
Node<E> p = head.next;
while (p != null) {
if (o.equals(p.item)) {
removed = true;
break;
}
trail = p;
p = p.next;
}
if (removed) {
p.item = null;
trail.next = p.next;
if (last == p)
last = trail;
if (count.getAndDecrement() == capacity)
notFull.signalAll();
}
} finally {
fullyUnlock();
}
return removed;
}
……
}
--------------------------------------------------------

http://hi.baidu.com/yao1111yao/item/eb1ea344bc2bc52511ee1e6d

快乐成长

每天进步一点点

来源：互联网

Queue

Related Posts:

发表评论 取消回复

发表评论取消回复