我们都知道Queue是一种操作线性表，可以用数组、链表来实现。
Queue的特点是先进先出。
存在问题：
如果我们使用普通的数组来实现Queue，一定会存在一个问题，每次出对的时候，从数组的第一位开始取，那么只要出对完成，就会涉及到整个数组搬迁，会导致时间最差复杂度成为O(n) 。单纯从设计Queue来讲，我觉得是应该要避免这样的搬迁操作。
循环数组
我们都知道数组是一块连续的内存空间。但是循环数组，确实不多见。如何理解？
环形数组
如上图，是一个大神的画的，看着挺复杂的，实现起来并没有那么复杂，只要抓住核心tail和head 即可，还有一个很重要的一个算法 (tail + 1) % n ，如下代码。
public class CircularQueue {
// 数组：items，数组大小：n
private String[] items;
private int n = 0;
// head表示队头下标，tail表示队尾下标
private int head = 0;
private int tail = 0;
// 申请一个大小为capacity的数组
public CircularQueue(int capacity) {
items = new String[capacity];
n = capacity;
}
// 入队
public boolean enqueue(String item) {
// 队列满了
if ((tail + 1) % n == head) return false;
items[tail] = item;
tail = (tail + 1) % n;
return true;
}
// 出队
public String dequeue() {
// 如果head == tail 表示队列为空
if (head == tail) return null;
String ret = items[head];
head = (head + 1) % n;
return ret;
}
}
这是一段使用循环数组实现简单Queue的代码，这样实现完全解决了上面的问题，出对的时候时间复杂度从O(n) 降成O(1)。
如果仔细看代码你会发现，在执行到临界点的时候，即队列满了，会存在有一个位置为空，内存如此宝贵怎么能忍。
有个简单操作，show you code。
public class CircularQueue {
// 数组：items，数组大小：n
private String[] items;
private int n = 0;
// 定义数组的长度
private int size = 0;
// head表示队头下标，tail表示队尾下标
private int head = 0;
private int tail = 0;
// 申请一个大小为capacity的数组
public CircularQueue(int capacity) {
items = new String[capacity];
n = capacity;
}
// 入队
public boolean enqueue(String item) {
if (size == n) {
return false;
}
// 队列满了
items[tail] = item;
tail = (tail + 1) % n;
size++;
return true;
}
// 出队
public String dequeue() {
// 如果head == tail 表示队列为空
if (size == 0) {
return null;
}
String ret = items[head];
head = (head + 1) % n;
size--;
return ret;
}
}
上述简单介绍了循环队列，设计确认比较巧妙，对于一些底层的编码，什么都要节省，如果有方式方法能节省就节省吧，毕竟现在我们都是天天把节省成本挂在嘴边。
重点看下ArrayBlockingQueue
之前一直用ArrayBlockingQueue，也看过几次源码，看了也忘记了。一直没有抓住核心点。这次看到了循环数组突然对ArrayBlockingQueue有些好奇。
是不是也是循环数组的方式来实现Queue
我们看下ArrayBlockingQueue源码，其他方法可以先不着急去看，先去看核心的方法enqueue和dequeue
/**
* Inserts element at current put position, advances, and signals.
* Call only when holding lock.
*/
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}
/**
* Extracts element at current take position, advances, and signals.
* Call only when holding lock.
*/
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
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.signal();
return x;
}
从上述代码来看，涉及到几个成员变量和ArrayBlockingQueue构造函数
/** The queued items */
final Object[] items;
/** 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;
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
/**
* Creates an {@code ArrayBlockingQueue} with the given (fixed)
* capacity and the specified access policy.
*
* @param capacity the capacity of this queue
* @param fair if {@code true} then queue accesses for threads blocked
* on insertion or removal, are processed in FIFO order;
* if {@code false} the access order is unspecified.
* @throws IllegalArgumentException if {@code capacity < 1}
*/
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();
}
其中takeIndex、putIndex、count 都是int 类型，默认为0。
notEmpty 表示队列为空的时候，出对操作的线程状态变成等待状态，并释放锁，入队操作的时候，会通知唤醒等待时间最长的线程。
notFull 表示队列为满的时候，入队操作线程状态变成等待状态，并释放锁，出对操作的时候，会通知唤醒等待时间最长的线程。
从ArrayBlockingQueue 中enqueue 和dequeue 源码看很容易理解使用的就是循环数组来实现的，如下代码
if (++putIndex == items.length)
putIndex = 0;
if (++takeIndex == items.length)
takeIndex = 0;
光看enqueue 和dequeue 源码发现其实整个逻辑非常不严谨，不严谨也没关系，主要看他们俩都是私有方法。更何况有个核心成员变量 count还没有涉及到。
ArrayBlockingQueue 对外提供的出对和入对操作都是成对，我们一一看来
/**
* Inserts the specified element at the tail of this queue if it is
* possible to do so immediately without exceeding the queue's capacity,
* returning {@code true} upon success and {@code false} if this queue
* is full. This method is generally preferable to method {@link #add},
* which can fail to insert an element only by throwing an exception.
*
* @throws NullPointerException if the specified element is null
*/
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 E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
offer和poll方法中 通过count 来 判断队列是否满了，或者是否为空队列，前提都加了Lock，为了保证线程安全，出入对都统一锁了。
到这里可能会有疑问阻塞去哪了？
我们再看另外一对的出入对方法
/**
* Inserts the specified element at the tail of this queue, waiting
* for space to become available if the queue is full.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
enqueue(e);
} 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();
}
}
notFull.await() 和 notEmpty.await() 使用上了，表示这俩个操作是阻塞的，很容易理解。
这里面有一个细节点需要注意
while (count == items.length)
notFull.await();
while (count == 0)
notEmpty.await();
我刚开始看的时候，我也有疑问为什么使用while而非if判断，其实是做了双重保证，防止过早或者意外通知唤醒。
再看下最后一对
/**
* Inserts the specified element at the tail of this queue, waiting
* up to the specified wait time for space to become available if
* the queue is full.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
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();
}
}
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();
}
}
理解了上面俩对出入对的方法，看到这对也是比较容易理解，加了阻塞时间而已。
总结
- 在我们经常使用基于数组的队列，里面也是使用循环数组的思想
- 循环数组的核心，就是一个圆环，通过tail 和head指针来定义出入对的索引。
- 循环数组的实现，有很多种，只要理解了思想，怎么实现就都可以。
- ArrayBlockingQueue 是一种阻塞队列，也提供非阻塞的线程安全出入对方法。
- ArrayBlockingQueue 阻塞实现是通过lock Condition来实现的。
今天是辛丑年最后一天，除夕，回不了自己的家乡，只能待在杭城。
在此祝所有的亲朋好友除夕安康，来年虎虎生威！
网友评论