MENU

常见队列模型

2019 年 07 月 13 日 • 阅读: 149 • 算法

什么是队列

和栈一样,队列 也是一种受限线性表,该模型是从现实生活中的排队抽象而来。想象一下,在车站排队买票时,先来的先买,后来的只能站末尾,不允许插队。先进先出,这就是队列的最大特征。对于栈,其只支持两个基本操作,入栈和出栈。队列也是如此,只支持入队 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,则数据搬移的平均次数为

$$\sum_{i=1}^{n-1}\frac{n-i}{n-1}=n/2$$

所以平均时间复杂度 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 机制便可实现非常高效的并发队列,这也是循环队列比链式队列应用更加广泛的原因。

实际上,对于大部分资源有限的场景,当没用空闲资源时,基本可以通过队列实现请求排队,如数据库连接池等。此外队列大小应根据资源数量设定,太大会导致排队等待时间过长,太小又会导致资源无法充分利用。

参考文献

最后编辑于: 2019 年 08 月 02 日