阻塞队列在并发编程非常常用，被广泛使用在“生产者-消费者”问题中。接下来两篇文章就来详细介绍阻塞队列。本文是阻塞队列上篇。
- 介绍
- 基本操作
- 应用
- 常用阻塞队列及源码 4.1 ArrayBlockingQueue 4.2 LinkedBlockingQueue 4.3 SynchronousQueue 4.4 PriorityBlockingQueue 4.5 DelayQueue
1. 介绍
阻塞队列（BlockingQueue）是一个比普通队列多出两个附加操作的队列。两个操作分别是：
- 在队列为空时，获取元素的线程会等待队列变为非空。
- 当队列满时，存储元素的线程会等待队列可用。
阻塞队列（BlockingQueue）被广泛使用在“生产者-消费者”问题中。当队列容器已满，生产者线程会被阻塞，直到队列未满；当队列容器为空时，消费者线程会被阻塞，直至队列非空时为止。这样可以对各个模块的业务功能进行解耦，生产者将“生产”出来的数据放置在数据容器中，而消费者仅仅只需要在“数据容器”中进行获取数据即可，这样生产者线程和消费者线程就能够进行解耦，只专注于自己的业务功能即可。
2. 基本操作
BlockingQueue基本操作如下：
代码语言：javascript复制// 插入元素
add(E e) ：往队列插入数据，当队列满时，插入元素时会抛出IllegalStateException异常；
offer(E e)：当往队列插入数据时，插入成功返回true，否则则返回false。当队列满时不会抛出异常；
// 删除元素
remove(Object o)：从队列中删除数据，成功则返回true，否则为false
poll：删除数据，当队列为空时，返回null；
// 查看元素
element：获取队头元素，如果队列为空时则抛出NoSuchElementException异常；
peek：获取队头元素，如果队列为空则抛出NoSuchElementException异常
// 插入数据：
put()：当阻塞队列容量已经满时，往阻塞队列插入数据的线程会被阻塞，直至阻塞队列已经有空余的容量可供使用；
offer(E e, long timeout, TimeUnit unit)：若阻塞队列已经满时，同样会阻塞插入数据的线程，直至阻塞队列已经有空余的地方，与put方法不同的是，该方法会有一个超时时间，若超过当前给定的超时时间，插入数据的线程会退出；
// 删除数据：
take()：当阻塞队列为空时，获取队头数据的线程会被阻塞；
poll(long timeout, TimeUnit unit)：当阻塞队列为空时，获取数据的线程会被阻塞，另外，如果被阻塞的线程超过了给定的时长，该线程会退出
put(e) 和 take() 是BlockingQueue的核心方法，也是我们比较关注的。
3. 应用
使用普通队列实现生产者-消费者模式，代码如下：
代码语言：javascript复制public class BlockingDemo {
static LinkedList<Integer> queue = new LinkedList<Integer>();
static int maxSize = 5;
public static void main(String[] args) throws Exception {
new Thread("生产者") {
public void run() {
while (true) {
synchronized (queue) {
while (queue.size() >= maxSize) {
try {
System.out.println("队列满了。。。");
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.addLast(1);
queue.notify();
System.out.println("队列中添加了一个元素, size=" queue.size());
}
}
};
}.start();
new Thread("消费者") {
public void run() {
while (true) {
synchronized (queue) {
while (queue.size() <= 0) {
try {
System.out.println("队列空了。。。");
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.removeFirst();
queue.notify();
System.out.println("队列中删除了一个元素, size=" queue.size());
}
}
};
}.start();
}
}
控制台输出：
代码语言：javascript复制队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中删除了一个元素, size=3
队列中删除了一个元素, size=2
队列中删除了一个元素, size=1
队列中删除了一个元素, size=0
队列空了。。。
队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列满了。。。
队列中删除了一个元素, size=4
队列中删除了一个元素, size=3
队列中删除了一个元素, size=2
队列中删除了一个元素, size=1
队列中删除了一个元素, size=0
队列空了。。。
队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列满了。。。
...部分省略...
使用阻塞队列实现生产者消费者模式：
代码语言：javascript复制public class Test {
static ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(5);
public static void main(String[] args) throws Exception {
new Thread("生产者") {
public void run() {
while (true) {
try {
queue.put(1);
System.out.println("队列中添加了一个元素, size=" queue.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
}.start();
Thread.sleep(500);
new Thread("消费者") {
public void run() {
while (true) {
try {
queue.take();
System.out.println("队列中删除了一个元素, size=" queue.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
}.start();
}
}
输出结果如下：
代码语言：javascript复制队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
...部分省略...
4. 常用阻塞队列
4.1 ArrayBlockingQueue
- ArrayBlockingQueue是由数组实现的有界队列，通过ReentrantLock锁保证队列数据的安全性，通过ReentrantLock的条件Condition是实现阻塞。
- 添加元素时，如果队列满了不能添加元素，就将添加元素的线程阻塞并加入notFull条件队列；当成功删除元素后，队列就可以添加元素了，唤醒notFull条件队列中阻塞的线程，添加元素。
- 删除元素时，如果队列空了不能删除元素，就将删除元素的线程阻塞并加入notEmpty条件队列；当成功添加元素后，队列就可以删除元素了，唤醒notEmpty条件队列中阻塞的线程，删除元素。
类结构
ArrayBlockingQueue是由数组实现的有界队列，通过ReentrantLock锁保证队列数据的安全性，通过ReentrantLock的条件Condition是实现阻塞。
队列创建时，确定队列大小和是否公平。
源码：
代码语言：javascript复制public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
final Object[] items;// 用于存放元素的数组
int takeIndex;// 下一次读取操作的位置
int putIndex;// 下一次写入操作的位置
int count;// 队列中的元素数量
// 通过lock及其两个条件notEmpty、notFull控制阻塞
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
// 创建队列时，确定队列大小和是否公平
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity];
lock = new ReentrantLock(fair);
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
}
put()
- 获取锁lock
- 队列满时，将当前线程加入notFull条件队列阻塞；当有元素出队时，队列就不满了，可以让元素入队了，此时会唤醒notFull条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后将元素入队。
- 入队：入队成功之后，唤醒notEmpty条件队列中阻塞的线程，让其元素出队
- 释放锁lock
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
final Object[] items;// 用于存放元素的数组
int takeIndex;// 下一次读取操作的位置
int putIndex;// 下一次写入操作的位置
int count;// 队列中的元素数量
// 通过lock及其两个条件notEmpty、notFull控制阻塞
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
}
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();// 获取lock锁
try {
/*
* 队列满时，将当前线程加入notFull条件队列阻塞；
* 当有元素出队时，队列就不满了，可以让元素入队了，
* 此时会唤醒notFull条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后将元素入队。
*/
while (count == items.length)
notFull.await();
enqueue(e);// 入队
} finally {
lock.unlock();// 解锁
}
}
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
if ( putIndex == items.length)
putIndex = 0;
count ;
// 入队成功之后，唤醒notEmpty条件队列中阻塞的线程，让其元素出队
notEmpty.signal();
}
take()
- 获取锁lock
- 队列空时，将当前线程加入notEmpty条件队列阻塞；当有元素入队时，队列不为空了就可以take出元素，此时会唤醒notEmpty条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后执行出队操作。
- 出队：出队成功，唤醒notFull条件队列中阻塞的线程，让其元素入队
- 释放锁lock
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();// 获取lock锁
try {
/*
* 队列空时，将当前线程加入notEmpty条件队列阻塞；
* 当有元素入队时，队列不为空了就可以take出元素，
* 此时会唤醒notEmpty条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后执行出队操作。
*/
while (count == 0)
notEmpty.await();
return dequeue();// 出队
} finally {
lock.unlock();// 解锁
}
}
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条件队列中阻塞的线程，让其元素入队
notFull.signal();
return x;
}
4.2 LinkedBlockingQueue
- LinkedBlockingQueue用链表实现的有界阻塞队列。（不设置容量，默认为Integer.MAX_VALUE）
- 锁takeLock保证删除数据的安全性，队列为空时读操作线程阻塞并加入takeLock锁的notEmpty条件等待队列。
- 锁putLock保证添加数据的安全性，队列满时写操作线程阻塞并加入putLock锁的notFull条件等待队列。
- ArrayBlockingQueue的读写使用同一个锁来保证数据安全。LinkedBlockingQueue的读写分别用不同的锁来保证数据安全，采用不同的锁可以使读线程和写线程并发执行，提高了吞吐量，但也增加了编程的复杂度。
类结构
LinkedBlockingQueue用链表实现的有界阻塞队列。（不设置容量，默认为Integer.MAX_VALUE）
锁takeLock保证删除数据的安全性，队列为空时读操作线程阻塞并加入takeLock锁的notEmpty条件等待队列。
锁putLock保证添加数据的安全性，队列满时写操作线程阻塞并加入putLock锁的notFull条件等待队列。
代码语言：javascript复制// 节点类 单向链表
static class Node<E> {
E item;
Node<E> next;
Node(E x) { item = x; }
}
private final int capacity;// 队列容量 不设置默认为Integer.MAX_VALUE
private final AtomicInteger count = new AtomicInteger(0);// 队列中的元素数量
private transient Node<E> head;// 队头
private transient Node<E> last;// 队尾
// take, poll, peek 等读操作的方法需要获取到这个锁
private final ReentrantLock takeLock = new ReentrantLock();
// 如果读操作的时候队列是空的，加入notEmpty等待队列
private final Condition notEmpty = takeLock.newCondition();
// put, offer 等写操作的方法需要获取到这个锁
private final ReentrantLock putLock = new ReentrantLock();
// 如果写操作的时候队列是满的，加入notFull等待队列
private final Condition notFull = putLock.newCondition();
// 有界队列， 不设置容量，默认为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);
}
put(E e)
- 获取putLock锁
- 如果队列满，当前线程阻塞并加入notFull条件等待队列
- 入队
- 这个元素入队成功后，队列还没有满，唤醒notFull队列中等待添加元素的线程。（因为添加元素和删除元素不是用的同一个锁导致会有这种情况发生）
- 释放掉putLock锁
- 如果添加元素前队列为空，可能会有读线程阻塞，所以在这个元素入队后，就唤醒阻塞的读线程
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();// 获取putLock锁
try {
// 如果队列满，当前线程阻塞并加入notFull条件等待队列
while (count.get() == capacity) {
notFull.await();
}
enqueue(node);// 入队
c = count.getAndIncrement();// count 原子加 1，注意：这里返回的c是原来的值，并不是加1后的值。
/*
* 这个元素入队成功后，队列还没有满，notFull.signal() 唤醒notFull队列中等待添加元素的线程。
* 为什么队列还没有满，但是添加元素线程却在阻塞状态呢？
* 因为添加元素和删除元素不是用的同一个锁，所以添加元素和删除元素是可以同时进行的。
* 当添加元素时发现队列满了，线程阻塞。此时另一个线程执行删除操作，队列又不满了。
* 于是出现了这个情况：队列还没有满，但是添加元素线程却在阻塞状态。
*/
if (c 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();// 入队后，释放掉 putLock
}
/*
* c == 0表示队列在这个元素入队前是空的，队列为空时可能会有读线程阻塞
* 所以在这个元素入队后，就唤醒阻塞的读线程
*/
if (c == 0)
signalNotEmpty();
}
/**
* 队列尾部插入元素
*/
private void enqueue(Node<E> node) {
last = last.next = node;
}
/**
* 唤醒读线程
*/
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
take()
- 获取takeLock锁
- 如果队列空，当前线程阻塞并加入notEmpty条件等待队列
- 出队
- 这个元素出队成功后，队列还有元素，唤醒notEmpty队列中等待删除元素的线程。（因为添加元素和删除元素不是用的同一个锁导致会有这种情况发生）
- 释放takeLock锁
- 如果添加元素前队列是满的，可能有写线程阻塞等待，所以唤醒写线程
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();// 获取锁takeLock
try {
// 如果队列为空，当前线程阻塞并加入notEmpty条件等待队列
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();// 出队
c = count.getAndDecrement();// count 进行原子减 1,注意：这里返回的c是原来的值，并不是减1后的值。
/*
* 这个元素出队成功后，队列还没有满，notEmpty.signal() 唤醒notEmpty队列中等待删除元素的线程。
* 当队列中还有元素时，为什么会有读线程在阻塞呢？
* 因为添加元素和删除元素不是用的同一个锁，所以添加元素和删除元素是可以同时进行的。
* 当删除元素时发现队列空了，线程阻塞。此时另一个线程执行添加操作，队列又不空了。
* 于是出现了这个情况：当队列中还有元素时，会有读线程在阻塞状态。
*/
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();// 释放锁takeLock
}
// c == capacity表示删除元素之前队列是满的，队列满时可能有写线程阻塞等待，所以唤醒写线程
if (c == capacity)
signalNotFull();
return x;
}
// 出队
private E dequeue() {
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
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();
}
}
