写在前面
本文一起来看下Java中阻塞队列相关内容。
1：队列
队列是一种先进先出FIFO的数据结构，是一种线性数据结构，一般有两种实现方式，一种是基于数组实现，另外一种是基于链表实现。接下来分别来看下。
1.1：基于数组实现
基于数组实现的队列叫做顺序队列，比较重要的点如下：
1：使用一个指针head指向下一个可以出队的位置。
2：使用一个指针tail作为下一个可以入队的位置。
3：队列空条件，head和tail重合。
4：队列满条件，tail的下一个位置是head。
如下图是从一个队列空到满的过程示意图：
如下图是出队列到队列空的过程示意图：
程序实现如下：
// 基于数组数据结构实现的队列（顺序队列）
public class ArrayQueue {
// 数组数组
private String[] items;
// 队列的大小
private int n;
// 队列头下标，下标所指为下一个可出队列元素
private int head;
// 队列尾下标，下一个所指为下一个可入队元素
private int tail;
// 构造函数，构造大小为capacity的队列
public ArrayQueue(int capacity) {
items = new String[capacity];
this.n = capacity;
head = 0;
tail = 0;
}
// 入队列操作
public boolean enqueue(String item) {
// 如果head和tail重合，并且head所指元素当前不为空则队列满
if (isFull()) {
System.out.println("队列已满！");
return false;
}
// 入队列
items[tail] = item;
// 如果已经到达队尾，则tail下标设置设置为1
// tail = (tail + 1) == n ? 0 : ++tail;
tail = (tail + 1) % n;
System.out.println(item + "入队列！");
return true;
}
// 出队列
public String dequeue() {
// 如果head和tail重合并且head所指元素为空则队列已空
if(isEmpty()) {
System.out.println("队列已空！");
return null;
}
// 出队列
String dequeueItem = items[head];
//items[head] = null;
//head = (head + 1) == n ? 0 : ++head;
head = (head + 1) % n;
System.out.println(dequeueItem + "出队列！");
return dequeueItem;
}
// 队列是否已空
public boolean isEmpty() {
return head == tail;
}
// 队列是否已满,这种判断满的方式会以浪费一个存储空间为代价
// 一旦head和tail在队列满时也重合，判队列空和满代价将更大（维护额外额外变量或者出队则清空数据）
public boolean isFull() {
return head == (tail + 1) % n;
}
// 省略getter setter tostring
}
测试代码如下：
class FakeCls {
public static void arrayQueueTest() {
ArrayQueue arrayQueue = new ArrayQueue(4);
arrayQueue.enqueue("1");
arrayQueue.enqueue("2");
arrayQueue.enqueue("3");
arrayQueue.enqueue("4");
arrayQueue.enqueue("5");
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.enqueue("6");
arrayQueue.enqueue("7");
arrayQueue.dequeue();
arrayQueue.enqueue("8");
arrayQueue.enqueue("9");
arrayQueue.enqueue("10");
arrayQueue.enqueue("11");
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
}
}
运行如下：
1入队列！
2入队列！
3入队列！
队列已满！
队列已满！
1出队列！
2出队列！
3出队列！
队列已空！
队列已空！
6入队列！
7入队列！
6出队列！
8入队列！
9入队列！
队列已满！
队列已满！
7出队列！
8出队列！
9出队列！
队列已空！
队列已空！
1.2：基于链表实现
基于链表实现的队列叫做链式队列，具体实现类似于数组，但是不同之处在于数组是通过下标来定位元素，而链表需要通过指针来定位元素，因此每个元素需要维护其上一个元素和下一个元素的指针，我们可以定义一个如下的节点类：
public class LinkQueueNode {
public LinkQueueNode next;
public LinkQueueNode prev;
public String data;
public LinkQueueNode() {
}
public LinkQueueNode(LinkQueueNode prev, LinkQueueNode next, String data) {
this.prev = prev;
this.next = next;
this.data = data;
}
// 省略getter setter tostring
}
然后定义如下的基于链表的队列实现类:
// 基于链表实现队列（链式队列）
public class LinkQueue {
public static final String HEAD_DATA = "head";
// 队列的第一个元素
private LinkQueueNode firstNode;
// 指向当前可出栈的元素的引用
private LinkQueueNode head;
// 指向当前可入栈的元素的引用
private LinkQueueNode tail;
// 队列的大小
private int n;
// 初始化大小为capacity的队列
public LinkQueue(int capacity) {
this.n = capacity;
LinkQueueNode curNode = null;
LinkQueueNode newNode = null;
for (int i = 0; i < capacity; i++) {
newNode = new LinkQueueNode(null, null, null);
if (i == 0) {
head = newNode;
tail = newNode;
firstNode = newNode;
} else {
curNode.setNext(newNode);
newNode.setPrev(curNode);
}
curNode = newNode;
}
}
// 入队
public boolean enqueue(String item) {
if (isFull()) {
System.out.println("队列已满！");
return false;
}
tail.data = item;
tail = (tail.next == null) ? firstNode : tail.next;
System.out.println(item + "入队！");
return true;
}
public boolean dequeue() {
if (isEmpty()) {
System.out.println("队列已空！");
return false;
}
String dequeueItem = head.data;
head.data = null;
head = (head.next == null) ? firstNode : head.next;
System.out.println(dequeueItem + "出队！");
return true;
}
public boolean isFull() {
return head == tail && head.getData() != null;
}
public boolean isEmpty() {
return head == tail && head.getData() == null;
}
// 省略getter setter tostring
}
测试代码如下：
class FakeCls {
private static void linkQueueTest() {
LinkQueue arrayQueue = new LinkQueue(4);
arrayQueue.enqueue("1");
arrayQueue.enqueue("2");
arrayQueue.enqueue("3");
arrayQueue.enqueue("4");
arrayQueue.enqueue("5");
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.enqueue("6");
arrayQueue.enqueue("7");
arrayQueue.dequeue();
arrayQueue.enqueue("8");
arrayQueue.enqueue("9");
arrayQueue.enqueue("10");
arrayQueue.enqueue("11");
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
arrayQueue.dequeue();
}
}
2：什么是阻塞队列?
以上不管是基于数组实现的队列还是基于链表实现的队列，当队列满时数据会无法插入，当队列空时会直接取不到数据，与此对应的如果是，当队列空时取队列数据线程等待直到队列有数据，当队列满时插入数据线程等待队列有空闲位置，具有这种行为的队列我们叫做是阻塞队列。接下来我们通过jdk提供的相关阻塞队列实现来一起看下。
2.1：BlockingQueue
这是在jdk的java.util.concurrent包中提供的一个接口，定义了阻塞队列相关的操作，其UML图如下：
可以看到其是java.util.Collection集合类的子接口，因此阻塞队列也是集合。接口源码如下：
// java.util.concurrent.BlockingQueue
// 在实现java.util.Queue定义的队列的基础支持了另外其他两个额外的操作，即，获取元素时等待队列变
// 为非空，存储一个元素时等待队列有可用空间，针对这种"等待队列有元素"/"等待队列有可用空间"的操作,
// 目前提供了四种行为，如下：
// throw exception,即当前无法满足目标操作时直接抛出异常，比如插入数据对应的方法是add，获取元素对应的remove
// special value,当前无法满足目标操作是返回特定值，比如插入数据对应的方法是offer，获取元素对应的poll
// blocks，阻塞等待直到目标操作被满足，比如插入元素对应的方法是put，获取元素对应的take
// times out，等待一定的时间来满足目标操作，否则超时，比如插入元素对应的元素是offer(e, time, unit)，获取元素对应的poll(time, unit)
// BlockingQueue不允许插入null元素，add，put，offer等都不允许，不允许的原因是null是作为失败的poll操作的返回值的
// BlockingQueue最初以用来设计使用在"生产者-消费者"场景中，但是同时还实现了java.util.Collection接口，因此
// 并不是绝对的一端入队，一端出队列，但是实际使用中我觉得还是当做一个标准的队列来用，不然会"乱套"。
// BlockingQueue是所有实现都是线程安全的，内部通过锁，等其他并发控制机制来实现自身的高效原子性控制。
// BlockingQueue内部并不天然支持维护一些注入close，shutdown等状态来表明不会有新元素插入进来，该类需求可在特定的实现类中提供，
// 但是接口中不提供相关的变量和方法。
public interface BlockingQueue<E> extends Queue<E> {
// 插入数据，如果是存在可用控件插入成功的话则返回true，否则返回java.lang.IllegalStateException，对于
// 长度有限的队列建议优先使用offer(obj)方法，该方法不会抛出异常，而是返回特定值
boolean add(E e);
// 插入元素，如果是插入成功则返回true，如果是因为容量限制插入失败，返回false，该方法
// 要优于使用add方法，因为add方法在插入元素失败时会抛出异常
boolean offer(E e);
// 插入元素，当无可用空间时会等待直到有可用空间
void put(E e) throws InterruptedException;
// 插入元素，当无可用空间时会等待指定的时长，最终成功返回true，失败返回false，不会抛出异常
boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException;
// 获取并删除队列头元素，如果是无可用元素，则阻塞等待直到有可用元素
E take() throws InterruptedException;
// 获取并且删除队列头元素，如果当前无可用元素则等待指定的时长，如果超时前有元素插入则获取对应的元素
// ，否则返回null，不抛出异常
E poll(long timeout, TimeUnit unit)
throws InterruptedException;
// 返回可用容量，如果是没有固有的（intrinsic [?n'tr?ns?k]）容量限制，则返回Integer.MAX_VALUE,
// 需要注意不能通过该方法来判断后续元素的插入是否会成功，因为可能并发多线程操作的情况，即可能有其他线程删除或者是
// 插入元素
int remainingCapacity();
// 删除一个元素
boolean remove(Object o);
// 是否包含某个元素
public boolean contains(Object o);
// 移除队列中的所有元素并添加到集合c中
int drainTo(Collection<? super E> c);
// 移除队列中指定数量的元素并添加到指定的集合c中
int drainTo(Collection<? super E> c, int maxElements);
}
2.2：PriorityBlockingQueue
带有优先级的队列，基于堆实现，这里是小顶堆，如下测试代码：
class FakeCls {
public static void main(String[] args) throws Exception {
PriorityBlockingQueue<Integer> priorityBlockingQueue = new PriorityBlockingQueue<>();
priorityBlockingQueue.put(23);
priorityBlockingQueue.put(12);
priorityBlockingQueue.put(54);
priorityBlockingQueue.put(42);
Integer curEle = null;
while ((curEle = priorityBlockingQueue.poll()) != null) {
System.out.println(curEle);
}
}
}
运行输出如下：
12
23
42
54
可以看到是按照从小到大的顺序输出的。
堆是这样的一种数据结构，i>=0 当ele(i)>ele(2i+1)并且ele(i)>ele(2i+2)时是大顶堆。当ele(i)<ele(2i+1)并且ele(i)<ele(2i+2)时是大顶堆。
2.3：DelayQueue
带有优先级和延迟时长的队列，只有超过了延迟时间数据才会被返回，加入到其中的元素必须实现java.util.concurrent.Delayed接口。比如购物场景，当用户将某商品加入购物车30分钟后还没有进行支付，可以短信通知用户提醒付款。如下测试代码：
/**
* compareTo 方法必须提供与 getDelay 方法一致的排序
*/
class MyDelayedTask implements Delayed {
private String name ;
private long start = System.currentTimeMillis();
private long time ;
public MyDelayedTask(String name,long time) {
this.name = name;
this.time = time;
}
/**
* 需要实现的接口，获得延迟时间 用过期时间-当前时间
* @param unit
* @return
*/
@Override
public long getDelay(TimeUnit unit) {
return unit.convert((start+time) - System.currentTimeMillis(),TimeUnit.MILLISECONDS);
}
/**
* 用于延迟队列内部比较排序 当前时间的延迟时间 - 比较对象的延迟时间
* @param o
* @return
*/
// 注意：该方法的实现按照规范要和getDelay保持一致，即按照getDelay的结果来进行比较，为什么这样？因为规范要求如此，遵守即可！！！
@Override
public int compareTo(Delayed o) {
MyDelayedTask o1 = (MyDelayedTask) o;
return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
}
@Override
public String toString() {
return "MyDelayedTask{" +
"name='" + name + '\'' +
", time=" + time +
'}';
}
}
public class TT {
private static DelayQueue delayQueue = new DelayQueue();
public static void main(String[] args) throws InterruptedException {
new Thread(new Runnable() {
@Override
public void run() {
delayQueue.offer(new MyDelayedTask("task1",10000));
delayQueue.offer(new MyDelayedTask("task2",3900));
delayQueue.offer(new MyDelayedTask("task3",1900));
delayQueue.offer(new MyDelayedTask("task4",5900));
delayQueue.offer(new MyDelayedTask("task5",6900));
delayQueue.offer(new MyDelayedTask("task6",7900));
delayQueue.offer(new MyDelayedTask("task7",4900));
}
}).start();
while (true) {
// 注意用take方法，不能用poll
Delayed take = delayQueue.take();
// Delayed take = delayQueue.poll();
System.out.println(take);
}
}
}
运行：
MyDelayedTask{name='task3', time=1900}
MyDelayedTask{name='task2', time=3900}
MyDelayedTask{name='task7', time=4900}
MyDelayedTask{name='task4', time=5900}
MyDelayedTask{name='task5', time=6900}
MyDelayedTask{name='task6', time=7900}
2.3.1：Delayed
接口``java.util.concurrent.Delayed`，源码如下：
// java.util.concurrent.Delayed
// 标记对象在经过一定的延迟时间之后才能被处理的接口，比如用在延迟队列中
// 注意：该接口的子类实现的compareTo方法必须和getDelay保持一致的顺序
public interface Delayed extends Comparable<Delayed> {
// 返回给定时间单位的剩余延迟时间，比如java.util.concurrent.TimeUnit.SECONDS,返回值是6，则代表延迟时间还有6秒
long getDelay(TimeUnit unit);
}
2.3.2：DelayQueue
- 插入元素offer
class FakeCls {
// java.util.concurrent.DelayQueue.offer(E)
// 插入指定的值（java.util.concurrent.Delayed接口子类）到延迟队列中，该值不能为null，否则会抛出java.lang.NullPoinerException
public boolean offer(E e) {
// 重入锁上锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
// private final PriorityQueue<E> q = new PriorityQueue<E>();优先级队列，底层基于堆实现
q.offer(e);
// 插入该元素e前队列为空，重置leader线程，通知available条件,可以消费数据
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
}
- 获取元素take
class FakeCls {
// java.util.concurrent.DelayQueue.take
// 获取并且删除队列头元素，如果是当前还没有延迟结束的元素，则等待直到有元素延迟结束
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
// 上锁
lock.lockInterruptibly();
try {
for (;;) {
// 获取头元素但不删除，用于处理当前队列空的场景（注意这是没有元素，而非没有延迟结束的元素）
E first = q.peek();
// 队列空，await，等待signal
if (first == null)
available.await();
else {
// 调用getDelay犯法获取当前队列头元素的延迟剩余时间
long delay = first.getDelay(NANOSECONDS);
// 延迟<0则说明可以获取，则调用poll获取元素，并返回
if (delay <= 0)
return q.poll();
first = null;
// 等待delay时长,继续for (;;)
if (leader != null)
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
}
- 获取元素poll
class FakeCls {
// java.util.concurrent.DelayQueue.poll()
// 获取队列头元素，如果是当前无元素或者是无延迟结束元素则返回null
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 获取队列头元素,只返回不删除，仅用于判断
E first = q.peek();
// 头元素为null，或者是当前头元素getDelay结果大于0，即延迟还没有结束
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
// 获取头元素，并删除
return q.poll();
} finally {
lock.unlock();
}
}
}
2.4：ArrayBlockingQueue
基于数组实现的阻塞队列，构造函数源码如下：
class FakeCls {
// 队列元素
final Object[] items;
// 访问控制锁
final ReentrantLock lock;
// 获取元素等待信号量
private final Condition notEmpty;
// 插入元素等待信号量
private final Condition notFull;
/** items index for next take, poll, peek or remove */
// 当前可以出队的元素的位置
int takeIndex;
/** items index for next put, offer, or add */
// 当前可以入队的位置
int putIndex;
/** Number of elements in the queue */
// 队列中元素的个数
int count;
// 指定大小创建阻塞队列
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
// 指定大小，并设置公平策略
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();
}
// 直接初初始化数据
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
this(capacity, fair);
final ReentrantLock lock = this.lock;
lock.lock(); // Lock only for visibility, not mutual exclusion
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;
} finally {
lock.unlock();
}
}
}
2.4.1：添加元素put
该方法在无可用空间时会阻塞等待，源码如下：
class FakeCls {
// java.util.concurrent.ArrayBlockingQueue.put
// 入队指定的元素，若当前队列已满则等待队列有可用空间
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
// 2022年3月15日15:25:20
// 已满，则等待，此处是DCL
while (count == items.length)
notFull.await();
// 2022年3月15日15:27:34
enqueue(e);
} finally {
lock.unlock();
}
}
}
2022年3月15日15:25:20处是等待队列有可用空间，并且使用了DCL ,防止线程唤醒的一瞬间，其他线程已经抢先一步入队元素。2022年3月15日15:27:34处是入队方法，具体参考2.4.2：入队enqueue。
2.4.2：入队enqueue
源码如下：
class FakeCls {
// java.util.concurrent.ArrayBlockingQueue.enqueue
// 入队
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
// putIndex为可入队位置，因此元素到数组该位置即可
items[putIndex] = x;
// 如果是到达数组最大长度，则从头，即索引位置0开始
if (++putIndex == items.length)
putIndex = 0;
// 元素个数+1
count++;
// 唤醒等待读取数据线程
notEmpty.signal();
}
}
2.4.3：获取元素poll
该方法在有可用元素时获取元素，无可用元素是返回null（默认值），源码如下：
class FakeCls {
// java.util.concurrent.ArrayBlockingQueue.poll()
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 2022年3月15日15:51:48
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
}
2022年3月15日15:51:48处如果是当前无元素，返回null，否则调用dequeue方法获取元素，具体参考2.4.4：出队dequeue。
2.4.4：出队dequeue
源码如下：
class FakeCls {
// java.util.concurrent.ArrayBlockingQueue.dequeue
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
// takeIndex为下一个出队的元素位置，因此直接通过其获取元素
E x = (E) items[takeIndex];
// 置null
items[takeIndex] = null;
// 如果是到达数组最后，则从头，即0索引位置开始
if (++takeIndex == items.length)
takeIndex = 0;
// 减少队列元素个数
count--;
// 更新迭代器元素数据
if (itrs != null)
itrs.elementDequeued();
// 唤醒等待入队元素的线程
notFull.signal();
return x;
}
}
2.5：LinkedBlockingDeque
基于链表实现的阻塞队列，主要源码如下：
class FakeCls {
// 双端链表内部类
static final class Node<E> {
// 数据元素，为null说明元素已经被移除
E item;
// 以下三种值之一：
// - 真正前驱节点
// - 本节点，意味着真正的前驱结点是尾结点
// - null，意味着没有前驱节点
Node<E> prev;
/**
* 以下三种值之一:
* - 真正的后继节点
* - 当前节点，意味着后继节点是头结点
* - null, 意味着没有后继节点
*/
Node<E> next;
Node(E x) {
item = x;
}
}
// 指向第一个节点，即可出栈的元素
transient Node<E> first;
// 下一个可入队的节点
transient Node<E> last;
// 队列元素个数
private transient int count;
// dequeue:双端队列，最大容量
private final int capacity;
final ReentrantLock lock = new ReentrantLock();
// 读等待信号量
private final Condition notEmpty = lock.newCondition();
// 写等待信号量
private final Condition notFull = lock.newCondition();
}
该类在dubbo的集群容错策略中ForkingCluster 中使用到了，是一个非常典型的使用场景。
写在后面
参考文章列表：