ArrayBlockingQueue阻塞队列

ArrayBlockingQueue是一个由数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素。新元素插入到队列的尾部，队列检索操作则是从队列头部开始获得元素。

/** 数组存储 */

private final E[] items;

/** 队尾下标 */

private int takeIndex;

/** 队列头下标 */

private int putIndex;

/** 数组的长度 */

private int count;

• offer()方法，将元素插入到队列，如果队列满了就返回false。

public boolean offer(E e) {

if (e == null) throw new NullPointerException();

final ReentrantLock lock = this.lock;

//如果没有被其它线程持有，便获取。否则等待

lock.lock();

try {

if (count == items.length)

return false;

else {

insert(e);

return true;

}

} finally {

lock.unlock();

}

}

• put()方法，将指定的元素插入此队列的尾部，如果该队列已满，则等待可用的空间。

public void put(E e) throws InterruptedException {

if (e == null) throw new NullPointerException();

final E[] items = this.items;

final ReentrantLock lock = this.lock;

/**线程在请求lock并被阻塞时，如果被interrupt，则此线程会被唤醒并被要求处理InterruptedException*/

lock.lockInterruptibly();

try {

try {

while (count == items.length)

notFull.await();

} catch (InterruptedException ie) {

//唤醒一个等待线程。

notFull.signal();

throw ie;

}

insert(e);

} finally {

lock.unlock();

}

}

• poll()方法，获取并移除此队列的头，如果此队列为空，则返回 null。

public E poll() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

if (count == 0)

return null;

E x = extract();

return x;

} finally {

lock.unlock();

}

}

• take()方法，获取并移除此队列的头部，在元素变得可用之前一直等待。

public E take() throws InterruptedException {

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

try {

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

}

}

• 使用阻塞队列实现生产者、消费者。

public class BlockQueueTest {

ArrayBlockingQueue<Object> queue = new ArrayBlockingQueue<Object>(10);

public static void main(String[] args) {

BlockQueueTest test = new BlockQueueTest();

Producer producer = test.new Producer();

Consumer consumer = test.new Consumer();

producer.start();

consumer.start();

}

class Consumer extends Thread {

public void run() {

consume();

}

private void consume() {

while (true) {

try {

queue.take();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

class Producer extends Thread {

public void run() {

produce();

}

private void produce() {

while (true) {

try {

queue.put(1);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

}