问题描述

阻塞队列是指当队列已满时，调用入队方法（enqueue）的线程会被阻塞，无法继续添加新元素；同样地，当队列为空时，调用出队方法（dequeue）的线程也会被阻塞。此外，当队列中有空闲空间时，应通知正在等待入队的线程；当队列中新增可用元素时，应通知正在等待出队的线程。

解决方案

我们的队列将在构造函数中传入一个固定容量，使用数组作为底层数据结构来存储队列元素。我们将为阻塞队列类提供 enqueue 和 dequeue 两个公共接口，并维护 head（头指针）和 tail（尾指针）用于标识队列的前端和后端，同时用一个 size 变量实时记录当前队列中的元素数量。基于这些设计，阻塞队列类的基本结构如下所示：

1public class BlockingQueue<T> {
2
3    T[] array;
4    int size = 0;
5    int capacity;
6    int head = 0;
7    int tail = 0;
8
9    public BlockingQueue(int capacity) {
10        array = (T[]) new Object[capacity];
11        this.capacity = capacity;
12    }
13
14    public void enqueue(T item) {
15    }
16
17    public T dequeue() {
18    }
19}

我们先从 enqueue 方法开始实现。如果当前队列大小 size == capacity，说明队列已满，需要阻塞调用者。我们可以通过在 while 循环中调用 wait() 方法实现阻塞，循环条件为队列大小等于最大容量。一旦其他线程执行了出队操作，该条件将变为 false，从而退出循环。

需要注意的是，每次检查 size 变量时，都必须确保没有其他线程正在修改它。为此，我们可以使用 synchronized 关键字，保证同一时刻只有一个线程能调用 enqueue 或 dequeue 方法。

另外，随着队列不断增长，尾指针可能会到达底层数组的末尾，此时需要将 tail 重置为 0。由于我们只在 size < capacity 时才允许入队，因此可以确保 tail 不会覆盖已有元素。

1    public synchronized void enqueue(T item) throws InterruptedException {
2
3        // 等待队列有空位
4        while (size == capacity) {
5            wait();
6        }
7
8        // 如果 tail 已经到达数组末尾，则重置为 0
9        if (tail == capacity) {
10            tail = 0;
11        }
12
13        // 将元素放入数组
14        array[tail] = item;
15        size++;
16        tail++;
17
18        // 通知所有等待的线程：队列状态已改变
19        // 可能有消费者线程在等待队列非空
20        notifyAll(); 
21    }

注意最后调用了 notifyAll() 方法。因为我们刚刚向队列中添加了一个元素，可能有消费者线程正阻塞在 dequeue 方法中等待元素出现，因此必须发送信号唤醒这些等待线程。

如果没有线程在等待，这个通知将被忽略，不会影响类的正确性。这不是“丢失信号”（missed signal）问题，因为队列状态已经改变。

接下来设计 dequeue 方法。与 enqueue 类似，当队列为空（即 size == 0）时，需要阻塞调用者。

当 head 指针超出数组范围时，需将其重置为 0。同时要减少 size 的值，因为出队后元素数量减少。

最后别忘了调用 notifyAll()，因为如果队列之前是满的，可能有生产者线程正阻塞在 enqueue 方法中等待空间释放。代码逻辑如下：

1    public synchronized T dequeue() throws InterruptedException {
2
3        T item = null;
4
5        // 等待至少有一个元素入队
6        while (size == 0) {
7            wait();
8        }
9
10        // 如果 head 超出数组范围，则重置为 0
11        if (head == capacity) {
12            head = 0;
13        }
14        
15        // 保存即将出队的元素引用，并将其位置置空
16        item = array[head];
17        array[head] = null;
18        head++;
19        size--;
20
21        // 唤醒可能阻塞在 enqueue 中的线程
22        notifyAll();
23
24        return item;
25    }

可以看出，dequeue 方法与 enqueue 方法是对称的。理论上我们可以省略第17、18行，直接写成：

return array[head-1];

但为了提高代码可读性，我们选择将其拆分为两行。

完整代码

完整的阻塞队列代码如下：

1class Demonstration {
2    public static void main( String args[] ) throws Exception{
3       final BlockingQueue<Integer> q = new BlockingQueue<Integer>(5);
4
5        Thread t1 = new Thread(new Runnable() {
6
7            @Override
8            public void run() {
9                try {
10                    for (int i = 0; i < 50; i++) {
11                        q.enqueue(new Integer(i));
12                        System.out.println("入队 " + i);
13                    }
14                } catch (InterruptedException ie) {
15
16                }
17            }
18        });
19
20        Thread t2 = new Thread(new Runnable() {
21
22            @Override
23            public void run() {
24                try {
25                    for (int i = 0; i < 25; i++) {
26                        System.out.println("线程2 出队: " + q.dequeue());
27                    }
28                } catch (InterruptedException ie) {
29
30                }
31            }
32        });
33
34        Thread t3 = new Thread(new Runnable() {
35
36            @Override
37            public void run() {
38                try {
39                    for (int i = 0; i < 25; i++) {
40                        System.out.println("线程3 出队: " + q.dequeue());
41                    }
42                } catch (InterruptedException ie) {
43
44                }
45            }
46        });
47
48        t1.start();
49        Thread.sleep(4000);
50        t2.start();
51
52        t2.join();
53
54        t3.start();
55        t1.join();
56        t3.join();
57    }
58}
59
60// 阻塞队列类
61class BlockingQueue<T> {
62
63    T[] array;
64    Object lock = new Object();
65    int size = 0;
66    int capacity;
67    int head = 0;
68    int tail = 0;
69
70    @SuppressWarnings("unchecked")
71    public BlockingQueue(int capacity) {
72        // 强制类型转换会产生警告
73        array = (T[]) new Object[capacity];
74        this.capacity = capacity;
75    }
76
77    public void enqueue(T item) throws InterruptedException {
78
79        synchronized (lock) {
80
81            while (size == capacity) {
82                lock.wait();
83            }
84
85            if (tail == capacity) {
86                tail = 0;
87            }
88
89            array[tail] = item;
90            size++;
91            tail++;
92            lock.notifyAll();
93        }
94    }
95
96    public T dequeue() throws InterruptedException {
97
98        T item = null;
99        synchronized (lock) {
100
101            while (size == 0) {
102                lock.wait();
103            }
104
105            if (head == capacity) {
106                head = 0;
107            }
108
109            item = array[head];
110            array[head] = null;
111            head++;
112            size--;
113
114            lock.notifyAll();
115        }
116
117        return item;
118    }
119}

本例中的测试用例创建了两个出队线程和一个入队线程。入队线程最初会填满队列并被阻塞，直到两个出队线程启动并从队列中取出元素。输出结果会在前五次入队之后，显示出交替进行的入队和出队操作。

进阶问题

在 enqueue() 和 dequeue() 方法中，我们都使用了 notifyAll() 而不是 notify()。这一选择背后的原因至关重要。考虑这样一个场景：有两个生产者线程和一个消费者线程共同操作一个容量为1的队列。当其中一个生产者线程向队列添加元素后，另外两个线程可能都在条件变量上等待。如果此时调用 notify()，系统可能唤醒另一个生产者线程。而该线程醒来后发现队列仍为满状态，会再次进入等待，导致消费者线程无法被唤醒，从而引发死锁。使用 notifyAll() 可确保消费者线程也有机会被唤醒并继续执行。