BlockingQueue

时间 2019-12-05

原文原文链接

BlockingQueue介绍与经常使用方法

BlockingQueue是一个阻塞队列。在高并发场景是用得很是多的，在线程池中。若是运行线程数目大于核心线程数目时，也会尝试把新加入的线程放到一个BlockingQueue中去。队列的特性就是先进先出很容易理解，在Java里头它的实现类主要有下图的几种，其中最经常使用到的是ArrayBlockingQueue、LinkedBlockingQueue及SynchronousQueue这三种。html

它主要的方法有java

BlockingQueue的核心方法：
一、放入数据android

（1） add(object)数组

队列没满的话，放入成功。不然抛出异常。缓存

（2）offer(object):安全

表示若是可能的话,将object加到BlockingQueue里,即若是BlockingQueue能够容纳,则返回true,不然返回false.（本方法不阻塞当前执行方法的线程）
（3）offer(E o, long timeout, TimeUnit unit)并发

能够设定等待的时间，若是在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。
（4）put(object)app

把object加到BlockingQueue里,若是BlockQueue没有空间,则调用此方法的线程阻塞。直到BlockingQueue里面有空间再继续.
二、获取数据
（1）poll(time)ide

取走BlockingQueue里排在首位的对象,若不能当即取出,则能够等time参数规定的时间,取不到时返回null;
（2）poll(long timeout, TimeUnit unit)函数

从BlockingQueue取出一个队首的对象，若是在指定时间内，队列一旦有数据可取，则当即返回队列中的数据。不然知道时间超时尚未数据可取，返回失败。

（3）take()

取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入;
（4）drainTo()

一次性从BlockingQueue获取全部可用的数据对象（还能够指定获取数据的个数），经过该方法，能够提高获取数据效率；不须要屡次分批加锁或释放锁。

ArrayBlockingQueue

一个由数组支持的有界阻塞队列。它的本质是一个基于数组的BlockingQueue的实现。
它的容纳大小是固定的。此队列按 FIFO（先进先出）原则对元素进行排序。
队列的头部是在队列中存在时间最长的元素。队列的尾部是在队列中存在时间最短的元素。
新元素插入到队列的尾部，队列检索操做则是从队列头部开始得到元素。
这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。
一旦建立了这样的缓存区，就不能再增长其容量。
试图向已满队列中放入元素会致使放入操做受阻塞，直到BlockingQueue里有新的唤空间才会被醒继续操做；

试图从空队列中检索元素将致使相似阻塞，直到BlocingkQueue进了新货才会被唤醒。
此类支持对等待的生产者线程和使用者线程进行排序的可选公平策略。
默认状况下，不保证是这种排序。然而，经过在构造函数将公平性 (fairness) 设置为 true 而构造的队列容许按照 FIFO 顺序访问线程。
公平性一般会下降吞吐量，但也减小了可变性和避免了“不平衡性”。
此类及其迭代器实现了 Collection 和 Iterator 接口的全部可选方法。
注意1：它是有界阻塞队列。它是数组实现的，是一个典型的“有界缓存区”。数组大小在构造函数指定，并且今后之后不可改变。
注意2:是它线程安全的，是阻塞的，具体参考BlockingQueue的“注意4”。
注意3:不接受 null 元素
注意4：公平性 (fairness)能够在构造函数中指定。

Public Constructors
	ArrayBlockingQueue(int capacity) Creates an ArrayBlockingQueue with the given (fixed) capacity and default access policy.
	ArrayBlockingQueue(int capacity, boolean fair) Creates an ArrayBlockingQueue with the given (fixed) capacity and the specified access policy.
	ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) Creates an ArrayBlockingQueue with the given (fixed) capacity, the specified access policy and initially containing the elements of the given collection, added in traversal order of the collection's iterator.

若是为true，则按照 FIFO 顺序访问插入或移除时受阻塞线程的队列；若是为 false，则访问顺序是不肯定的。
  注意5:它实现了BlockingQueue接口。
  注意6：此类及其迭代器实现了 Collection 和 Iterator 接口的全部可选方法。
  注意7：其容量在构造函数中指定。容量不能够自动扩展，也没提供手动扩展的接口。

注意8：在JDK5/6中，LinkedBlockingQueue和ArrayBlocingQueue等对象的poll(long timeout, TimeUnit unit)存在内存泄露
Leak的对象是AbstractQueuedSynchronizer.Node，
据称JDK5会在Update12里Fix，JDK6会在Update2里Fix。

源码分析：

一个基本数组的阻塞队列。能够设置列队的大小。

它的基本原理实际仍是数组，只不过存、取、删时都要作队列是否满或空的判断。而后加锁访问。

[java] view plain copy

package java.util.concurrent;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.lang.ref.WeakReference;
import java.util.Spliterators;
import java.util.Spliterator;
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private static final long serialVersionUID = -817911632652898426L;
/** 真正存入数据的数组*/
final Object[] items;
/** take, poll, peek or remove的下一个索引 */
int takeIndex;
/** put, offer, or add的下一个索引 */
int putIndex;
/**队列中元素个数*/
int count;
/**可重入锁 */
final ReentrantLock lock;
/** 队列不为空的条件 */
private final Condition notEmpty;
/** 队列未满的条件 */
private final Condition notFull;
transient Itrs itrs = null;
/**
*当前元素个数-1
*/
final int dec(int i) {
return ((i == 0) ? items.length : i) - 1;
}
/**
* 返回对应索引上的元素
*/
@SuppressWarnings("unchecked")
final E itemAt(int i) {
return (E) items[i];
}
/**
* 非空检查
*
* @param v the element
*/
private static void checkNotNull(Object v) {
if (v == null)
throw new NullPointerException();
}
/**
* 元素放入队列，注意调用这个方法时都要先加锁
*
*/
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;//当前拥有元素个数加1
notEmpty.signal();//有一个元素加入成功，那确定队列不为空
}
/**
* 元素出队，注意调用这个方法时都要先加锁
*
*/
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--;/当前拥有元素个数减1
if (itrs != null)
itrs.elementDequeued();
notFull.signal();//有一个元素取出成功，那确定队列不满
return x;
}
/**
* 指定删除索引上的元素
*
*/
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 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();//有一个元素删除成功，那确定队列不满
}
/**
*
* 构造函数，设置队列的初始容量
*/
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
/**
* 构造函数。capacity设置数组大小，fair设置是否为公平锁
* capacity and the specified access policy.
*/
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity];
lock = new ReentrantLock(fair);//是否为公平锁，若是是的话，那么先到的线程先得到锁对象。
//不然，由操做系统调度由哪一个线程得到锁，通常为false，性能会比较高
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
/**
*构造函数，带有初始内容的队列
*/
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
this(capacity, fair);
final ReentrantLock lock = this.lock;
lock.lock(); //要给数组设置内容，先上锁
try {
int i = 0;
try {
for (E e : c) {
checkNotNull(e);
items[i++] = e;//依次拷贝内容
}
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
count = i;
putIndex = (i == capacity) ? 0 : i;//若是putIndex大于数组大小，那么从0从新开始
} finally {
lock.unlock();//最后必定要释放锁
}
}
/**
* 添加一个元素，其实super.add里面调用了offer方法
*/
public boolean add(E e) {
return super.add(e);
}
/**
*加入成功返回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();
}
}
/**
* 若是队列已满的话，就会等待
*/
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();//和lock()方法的区别是让它在阻塞时也可抛出异常跳出
try {
while (count == items.length)
notFull.await(); //这里就是阻塞了，要注意。若是运行到这里，那么它会释放上面的锁，一直等到notify
enqueue(e);
} finally {
lock.unlock();
}
}
/**
* 带有超时时间的插入方法，unit表示是按秒、分、时哪种
*/
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 (count == items.length) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);//带有超时等待的阻塞方法
}
enqueue(e);//入队
return true;
} finally {
lock.unlock();
}
}
//实现的方法，若是当前队列为空，返回null
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
//实现的方法，若是当前队列为空，一直阻塞
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();//队列为空，阻塞方法
return dequeue();
} finally {
lock.unlock();
}
}
//带有超时时间的取元素方法，不然返回Null
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);//超时等待
}
return dequeue();//取得元素
} finally {
lock.unlock();
}
}
//只是看一个队列最前面的元素，取出是不删除队列中的原来元素。队列为空时返回null
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // 队列为空时返回null
} finally {
lock.unlock();
}
}
/**
* 返回队列当前元素个数
*
*/
public int size() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
/**
* 返回当前队列再放入多少个元素就满队
*/
public int remainingCapacity() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return items.length - count;
} finally {
lock.unlock();
}
}
/**
* 从队列中删除一个元素的方法。删除成功返回true,不然返回false
*/
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;
} finally {
lock.unlock();
}
}
/**
* 是否包含一个元素
*/
public boolean contains(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]))
return true;
if (++i == items.length)
i = 0;
} while (i != putIndex);
}
return false;
} finally {
lock.unlock();
}
}
/**
* 清空队列
*
*/
public void clear() {
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int k = count;
if (k > 0) {
final int putIndex = this.putIndex;
int i = takeIndex;
do {
items[i] = null;
if (++i == items.length)
i = 0;
} while (i != putIndex);
takeIndex = putIndex;
count = 0;
if (itrs != null)
itrs.queueIsEmpty();
for (; k > 0 && lock.hasWaiters(notFull); k--)
notFull.signal();
}
} finally {
lock.unlock();
}
}
/**
* 取出全部元素到集合
*/
public int drainTo(Collection<? super E> c) {
return drainTo(c, Integer.MAX_VALUE);
}
/**
* 取出全部元素到集合
*/
public int drainTo(Collection<? super E> c, int maxElements) {
checkNotNull(c);
if (c == this)
throw new IllegalArgumentException();
if (maxElements <= 0)
return 0;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int n = Math.min(maxElements, count);
int take = takeIndex;
int i = 0;
try {
while (i < n) {
@SuppressWarnings("unchecked")
E x = (E) items[take];
c.add(x);
items[take] = null;
if (++take == items.length)
take = 0;
i++;
}
return n;
} finally {
// Restore invariants even if c.add() threw
if (i > 0) {
count -= i;
takeIndex = take;
if (itrs != null) {
if (count == 0)
itrs.queueIsEmpty();
else if (i > take)
itrs.takeIndexWrapped();
}
for (; i > 0 && lock.hasWaiters(notFull); i--)
notFull.signal();
}
}
} finally {
lock.unlock();
}
}
}

使用实例：

生产者-消费者模型

大量的实现ArrayBlockingQueue已经作掉了，包括判空，线程挂起等操做都封装在ArrayBlockingQueue中。生产者只须要关心生产，消费者只须要关心消费。而若是不使用ArrayBlockingQueue的话，具体的生产者还须要去通知消费者，还须要关心整个容器是否满了。从这里能够看出ArrayBlockingQueue是一种比较好的实现方式，高度的内聚。

Producer.java

[java] view plain copy

public class Producer implements Runnable{
//容器
private final ArrayBlockingQueue<Bread> queue;
public Producer(ArrayBlockingQueue<Bread> queue){
this.queue = queue;
}
/* (non-Javadoc)
* @see java.lang.Runnable#run()
*/
@Override
public void run() {
while(true){
produce();
}
}
public void produce(){
/**
* put()方法是若是容器满了的话就会把当前线程挂起
* offer()方法是容器若是满的话就会返回false。
*/
try {
Bread bread = new Bread();
queue.put(bread);
System.out.println("Producer:"+bread);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

Consumer.java

[java] view plain copy

public class Consumer implements Runnable{
//容器
private final ArrayBlockingQueue<Bread> queue;
public Consumer(ArrayBlockingQueue<Bread> queue){
this.queue = queue;
}
/* (non-Javadoc)
* @see java.lang.Runnable#run()
*/
@Override
public void run() {
while(true){
consume();
}
}
public void consume(){
/**
* take()方法和put()方法是对应的，从中拿一个数据，若是拿不到线程挂起
* poll()方法和offer()方法是对应的，从中拿一个数据，若是没有直接返回null
*/
try {
Bread bread = queue.take();
System.out.println("consumer:"+bread);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

Client.java

[java] view plain copy

public class Client {
/**
* @param args
*/
public static void main(String[] args) {
int capacity = 10;
ArrayBlockingQueue<Bread> queue = new ArrayBlockingQueue<Bread>(capacity);
new Thread(new Producer(queue)).start();
new Thread(new Producer(queue)).start();
new Thread(new Consumer(queue)).start();
new Thread(new Consumer(queue)).start();
new Thread(new Consumer(queue)).start();
}
}

参考：http://blog.csdn.net/evankaka/article/details/51706109

http://blog.csdn.net/hudashi/article/details/7076745