本文首发于 LOGI'S BLOG，由作者转载。
什么是队列
和栈一样，队列 也是一种受限线性表，该模型是从现实生活中的排队抽象而来。想象一下，在车站排队买票时，先来的先买，后来的只能站末尾，不允许插队。先进先出，这就是队列的最大特征。对于栈，其只支持两个基本操作，入栈和出栈。队列也是如此，只支持入队 enqueue，放数据到队尾；出队 dequeue，从队头取元素。
push pop
[ ] head
[ ]
[ ]
stack
------------------------------------
queue
head tail
dequeue<-[]<-[]<-[]<-[]<-[]<-enqueue
顺序队列
和栈一样，使用数组和链表都能实现队列，前者叫顺序队列，后者叫链式队列。观察下面的例子，我们便可总结出顺序队列的实现步骤。
1. 初始化一个容量为 capacity，不妨以 8 为例，的队列，
+ 此时头尾指针都指向数组第一个元素。
head
tail
[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]
0 1 2 3 4 5 6 7
2. 当 a,b,c,d,e,f,g,h 依次入队后，head 仍指向下标为 0 的位置，
+ tail 指向下标为 8 的位置，此时 tail=capacity，队列已满。
head tail
[a]<-[b]<-[c]<-[d]<-[e]<-[f]<-[g]<-[h]
0 1 2 3 4 5 6 7
3. a,b 出队，head 指向下标 2，tail 仍指向下标 8。
head tail
[ ]<-[ ]<-[c]<-[d]<-[e]<-[f]<-[g]<-[h]
0 1 2 3 4 5 6 7
4. 此时 i 入队，虽然队列仍有位置，但 tail=capacity，
+ 如果将它放到 tail 位置将导致数组越界。为了不影响出队操作的一致性，
+ 应该将下标从 head 到 tail-1 位置的元素整体迁移 head 个位置。相应地，
+ tail 也要前移 head 个位置，head 指针移动到 0 位置，最后将 i 入队。
head tail
[c]<-[d]<-[e]<-[f]<-[g]<-[h]<-[ ]<-[ ]
0 1 2 3 4 5 6 7
head tail
[c]<-[d]<-[e]<-[f]<-[g]<-[h]<-[i]<-[ ]
0 1 2 3 4 5 6 7
以下代码实现了上述过程，一般情况下，需要排队时说明资源有限，所以队列的大小一般是事先设定好的，不做扩容操作。
class ArrayBasedQueue<T> {
Object[] items;
int capacity;
int head;
int tail;
public ArrayBasedQueue(int capacity) {
this.capacity = capacity;
this.items = new Object[capacity];
}
// 入队
public boolean enqueue(T item) {
if (this.tail == this.capacity) {
if (this.head == 0) {
// 队列已满，直接返回
return false;
}
// 队列未满，先做数据搬移
for (int i = this.head; i < this.tail; i++) {
this.items[i] = this.items[i - this.head];
}
// tail、head 指针同样前移
this.tail -= this.head;
this.head = 0;
}
// 放到队尾并将 tail 指针后移
this.items[this.tail++] = item;
return true;
}
// 出队
public T dequeue() {
// 空队返回 null，否则返回队头并将 head 指针后移
return this.head == this.tail ? null : (T) items[head++];
}
}
复杂度分析：因为未使用额外空间，所以空间复杂度为 O(1)。出队直接返回数据，时间复杂度为 O(1)。入队时，当 tail=capacity 但队列未满时，需要搬移数据，搬移操作的时间复杂度由 head 的位置决定。最好情况是 head 在下标为 capacity-1 位置，仅需一次搬移，时间复杂度 O(1)。最坏情况是 head 在下标为 1 的位置，需要 capacity-1 次数据搬移，时间复杂度 O(n)。而 head 的下标可以是 1 到 capacity-1 中的任意数字，总共有 capacity-1 种情况，每种情况的概率相同，都为 1/(capacity-1)，而搬移次数为 capacity-head，记 capacity 为 n，head 下标为 i，则数据搬移的平均次数为
所以平均时间复杂度 O(n)。
入队的其他情况都无需搬移数据，时间复杂度 O(1)。平均来看，发生数据搬移的情况符合以下规律：当 head 下标不为 0 ，tail 为 head+1 时，前 capacity-(head+1) 次无需数据搬移，当 capacity=tail 时需要搬移 capacity-head 次，将其均摊到 capacity-head 次入队上，每次入队需搬移 1 次，时间复杂度 O(1)。
head
tail
[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]
0 1 2 3 4 5 6 7
head tail
[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]
0 1 2 3 4 5 6 7
但 head 的位置依赖入队，所以脱离实际算法无法给出精确的时间复杂度，但决大部分情况下，入队应该是 O(1) 操作，基于顺序队列的某些极端算法，可能会退化为 O(n)，需要根据具体算法分析。
链式队列
实际上，链表不存在下标越界问题，所以用它实现队列反而相对简单。我们仍然先观察示例，总结规律，随后完成编码。
1. 队列初始化，head 和 tail 指针都指向第一个结点，规定第一个结点不存储数据。
head
tail
[/]
2. 结点 a 入队，tail->next=newNode, tail=newNode。
head
tail
[/]->[a]
head tail
[/]->[a]
3. 结点 a 出队，head=head->next, return head->data。
head
tail
[/]->[/]
以下代码实现了上述过程，由于链表无需搬移数据，所以出入队的时间复杂度都是 O(1)，空间复杂度也是 O(1)，同时存储密度退化为 1/2。
class SinglyLinkedListBasedQueue<T> {
QueueNode head;
QueueNode tail;
int capacity;
int size;
public SinglyLinkedListBasedQueue(int capacity) {
this.capacity = capacity;
this.head = new QueueNode(null);
this.tail = head;
}
// 入队
public boolean enqueue(T item) {
if (this.size == this.capacity) {
// 队列已满，拒绝入队
return false;
}
this.tail.next = new QueueNode(item);
this.tail = this.tail.next;
this.size++;
return true;
}
// 出队
public T dequeue() {
if (this.head == this.tail) {
// 队空，无元素出队
return null;
}
this.head = this.head.next;
this.size--;
return (T) head.item;
}
}
class QueueNode {
QueueNode next;
Object item;
public QueueNode(Object item) {
this.item = item;
}
}
循环队列
上面用数组实现的顺序队列，入队时，有时会发生数据搬移，极端情况下时间复杂度为 O(n)，循环队列完美解决了该问题。顾名思义，顺序队列从头到尾是一条直线，循环队列是一个环，下面是它的示意图。
0 1 2 3
[ ]<-[ ]<-[ ]<-[ ]
↓ ↑
[ ]->[j]->[i]->[h]
7 6 5 4
tail head
上图的循环队列大小为 8，当前 head=4，tail=7。当有新元素 k 入队时，tail 不是更新为 8，而是变为 0。
tail
0 1 2 3
[ ]<-[ ]<-[ ]<-[ ]
↓ ↑
[k]->[j]->[i]->[h]
7 6 5 4
head
当再有一个元素 l 入队时，将其放入 0 位置，然后 tail+1 更新为 1。
tail
0 1 2 3
[l]<-[ ]<-[ ]<-[ ]
↓ ↑
[k]->[j]->[i]->[h]
7 6 5 4
head
由此可见，循环队列不会发生 tail=capacity 的情况，也就不需要数据搬移。但是，入队也要判断队列是否已满，出队要判断队列是否为空，在循环队列里，这两个操作该如何做到那？
初始化队列时，我们将 head 和 tail 都指向下标 0，要判断 head=tail 能否作为队空条件还需考虑以下情况：假设队列容量为 8，将 a,b,c,d,e,f,g 依次入队，每次 tail=++tail%capacity，入队完毕后 head 指向 0，tail 指向 7。再将他们依次出队，每次 head=++head%capacity，出队完毕后，head=tail=7。所以，将 head=tail 作为判空条件是可行的。
head
0 1 2 3
[a]<-[b]<-[c]<-[d]
↓ ↑
[ ]->[g]->[f]->[e]
7 6 5 4
tail
0 1 2 3
[ ]<-[ ]<-[ ]<-[ ]
↓ ↑
[ ]->[ ]->[ ]->[ ]
7 6 5 4
tail
head
再来寻找队满条件，接着上面的状态，head=tail=7，队列为空。依次将 a,b,c,d,e,f,g 入队，此时 head=7，tail=6。此时虽然数组仍有一个空间，但如果再入一个元素，head 将等于 tail，上面我们规定，head=tail 是队空条件，为了有所区分，我们只能规定 (tail+1)%capacity=head 为队满条件。理论上，你也可以用 (tail+2)%capacity=head 来判断队满，但那会浪费更多空间。所以 (tail+1)%capacity=head 是判断队满的最佳条件。
0 1 2 3
[b]<-[c]<-[d]<-[e]
↓ ↑
[a]->[ ]->[g]->[f]
7 6 5 4
head tail
以下代码实现了上述过程。
class ArrayBasedCircularQueue<T> {
Object[] items;
int capacity;
int head;
int tail;
public ArrayBasedCircularQueue(int capacity) {
this.capacity = capacity;
this.items = new Object[capacity];
}
// 入队
public boolean enqueue(T item) {
if ((this.tail + 1) % this.capacity == this.head) {
return false;
}
this.items[this.tail] = item;
this.tail = (this.tail + 1) % this.capacity;
return true;
}
// 出队
public T dequeue() {
if (this.head == this.tail) {
return null;
}
Object item = this.items[this.head];
this.head = (this.head + 1) % capacity;
return (T) item;
}
}
队列的常规应用
生产者-消费者问题
我们先介绍 阻塞对列，它其实就是在队列基础上增加了阻塞操作，具体来说，当队列为空时，从队头取数据的操作会被阻塞，直到队列有数据才返回；相应地，插入数据时，如果队列已满，该操作会阻塞，直到有空闲位置时再进行插入操作，然后返回。根据以上描述，我们很容易想到 生产者-消费者模型，这种模型可以有效协调生产和消费的速度，当生产者生产数据过快，消费者来不及消费时，存储队列很快就满了，此时生产者最好阻塞等待，以节省资源，或者启动更多消费者进行消费。
Customer Thread Take<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-Producer Thread Put
另一方面，当消费快于生产时，阻塞消费必然会降低系统效率，相反，我们应该启动更多生产者进行生产。
Customer Thread1 Take Producer Thread1 Put
Customer Thread2 Take<-[ ]<-[ ]<-[ ]<-[ ]<-[ ]<-Producer Thread2 Put
Customer Thread3 Take Producer Thread3 Put
高并发下的线程安全
线程安全队列又叫 并发队列，实现线程安全，最简单的方式就是直接在 enqueue 和 dequeue 上加锁，但大粒度加锁，同时仅允许一个线程存或取，势必会严重降低并发性。实际上，利用基于数组的循环队列，加上 CAS 机制便可实现非常高效的并发队列，这也是循环队列比链式队列应用更加广泛的原因。
实际上，对于大部分资源有限的场景，当没用空闲资源时，基本可以通过队列实现请求排队，如数据库连接池等。此外队列大小应根据资源数量设定，太大会导致排队等待时间过长，太小又会导致资源无法充分利用。
网友评论