手撸MQ消息队列——循环数组

原创

小忽悠

发布于 2024-09-14 14:38:15

570

发布于 2024-09-14 14:38:15

队列是咱们开发中经常使用到的一种数据结构，它与栈的结构类似。然而栈是后进先出，而队列是先进先出，说的专业一点就是FIFO。在生活中到处都可以找到队列的，最常见的就是排队，吃饭排队，上地铁排队，其他就不过多举例了。

队列的模型

在数据结构中，和排队这种场景最像的就是数组了，所以我们的队列就用数组去实现。在排队的过程中，有两个基本动作就是入队和出队，入队就是从队尾插入一个元素，而出队就是从队头移除一个元素。基本的模型我们可以画一个简图：

看了上面的模型，我们很容易想到使用数组去实现队列，

先定义一个数组，并确定数组的长度，我们暂定数组长度是5，而上面图中的长度是一样的；
再定义两个数组下标，front和tail，front是队头的下标，每一次出队的操作，我们直接取front下标的元素就可以了。tail是队尾的下标，每一次入队的操作，我们直接给tail下标的位置插入元素就可以了。

我们看一下具体的过程，初始状态是一个空的队列，

队头下标和队尾下标都是指向数组中的第0个元素，现在我们插入第一个元素“a”，如图：

数组的第0个元素赋值“a”，tail的下标+1,由指向第0个元素变为指向第1个元素。这些变化我们都要记住啊，后续在编程实现的过程中，每一个细节都不能忽略。然后我们再做一次出队操作：

第0个元素“a”在数组中移除了，并且front下标+1，指向第1个元素。

这些看起来不难实现啊，不就是给数组元素赋值，然后下标+1吗？但是我们想一想极端的情况，我们给数组的最后一个元素赋值后，数组的下标怎么办？

tail如果再+1，就超越了数组的长度了呀，这是明显的越界了。同样front如果取了数组中的最后一个元素，再+1，也会越界。这怎么办呢？

循环数组

我们最开始想到的方法，就是当tail下标到达数组的最后一个元素的时候，对数组进行扩容，数组的长度又5变为10。这种方法可行吗？如果一直做入队操作，那么数组会无限的扩容下去，占满磁盘空间，这是我们不想看到的。

另外一个方法，当front或tail指向数组最后一个元素时，再进行+1操作，我们将下标指向队列的开头，也就是第0个元素，形成一个循环，这就叫做循环数组。那么这里又引申出一个问题，我们的下标怎么计算呢？

数组的长度是5；
tail当前的下标是4，也就是数组的最后一个元素；
我们给最后一个元素赋值后，tail怎么由数组的最后一个下标4，变为数组的第一个下标0？

这里我们可以使用取模来解决：tail = (tail + 1) % mod，模（mod）就是我们的数组长度5，我们可以试一下，tail当前值是4，套入公式计算得到0，符合我们的需求。我们再看看其他的情况符不符合，假设tail当前值是1，套入公式计算得出2，也相当于是+1操作，没有问题的。只有当tail+1=5时，才会变为0，这是符合我们的条件的。那么我们实现队列的方法就选用循环数组，而且数组下标的计算方法也解决了。

队列的空与满

队列的空与满对入队和出队的操作是有影响的，当队列是满的状态时，我们不能进行入队操作，要等到队列中有空余位置才可以入队。同样当队列时空状态时，我们不能进行出队操作，因为此时队列中没有元素，要等到队列中有元素时，才能进行出队操作。那么我们怎么判断队列的空与满呢？

我们先看看队列空与满时的状态：

空时的状态就是队列的初始状态，front和tail的值是相等的。

满时的状态也是front==tail，我们得到的结论是，front==tail时，队列不是空就是满，那么到底是空还是满呢？这里我们要看看是什么操作导致的front==tail，如果是入队导致的front==tail，那么就是满；如果是出队导致的front==tail，那就是空。

手撸代码

好了，队列的模型以及基本的问题都解决了，我们就可以手撸代码了，我先把代码贴出来，然后再给大家讲解。

public class MyQueue<T> {

    //循环数组
    private T[] data;
    //数组长度
    private int size;
    //出队下标
    private int front =0;
    //入队下标
    private int tail = 0;
    //导致front==tail的原因，0：出队；1：入队
    private int flag = 0;

    //构造方法，定义队列的长度
    public MyQueue(int size) {
        this.size = size;
        data = (T[])new Object[size];
    }
    
    /**
     * 判断对队列是否满
     * @return
     */
    public boolean isFull() {
        return front == tail && flag == 1;
    }

    /**
     * 判断队列是否空
     * @return
     */
    public boolean isEmpty() {
        return front == tail && flag == 0;
    }

    /**
     * 入队操作
     * @param e
     * @return
     */
    public boolean add(T e) {
        if (isFull()) {
            throw new RuntimeException("队列已经满了");
        }
        data[tail] = e;
        tail = (tail + 1) % size;
        if (tail == front) {
            flag = 1;
        }

        return true;
    }

    /**
     * 出队操作
     * @return
     */
    public T poll() {
        if (isEmpty()) {
            throw new RuntimeException("队列中没有元素");
        }
        T rtnData = data[front];
        front = (front + 1) % size;
        if (front == tail) {
            flag = 0;
        }
        return rtnData;
    }
}

在类的开始，我们分别定义了，循环数组，数组的长度，入队下标，出队下标，还有一个非常重要的变量flag，它表示导致front==tail的原因，0代表出队，1代表入队。这里我们初始化为0，因为队列初始化的时候是空的，而且front==tail，这样我们判断isEmpty()的时候也是正确的。

接下来是构造方法，在构造方法中，我们定义了入参size，也就是队列的长度，其实就是我们循环数组的长度，并且对循环数组进行了初始化。

再下面就是判断队列空和满的方法，实现也非常的简单，就是依照上一小节的原理。

然后就是入队操作，入队操作要先判断队列是不是已经满了，如果满了，我们进行报错，不进行入队的操作。有的同学可能会说，这里应该等待，等待队列有空位了再去执行。这种说法是非常正确的，我们先把最基础的队列写完，后面还会再完善，大家不要着急。下面就是对循环数组的tail元素进行赋值，赋值后，使用我们的公式移动tail下标，tail到达最后一个元素时，通过公式计算，可以回到第0个元素。最后再判断一下，这个入队操作是不是导致了front==tail，如果导致了，就将flag置为1。

出队操作和入队操作类似，只不过是取值的步骤，这里不给大家详细解释了。

我们做个简单的测试吧，

 public static void main(String[] args) {
        MyQueue<String> myQueue = new MyQueue<>(5);
        System.out.println("isFull:"+myQueue.isFull()+" isEmpty:"+myQueue.isEmpty());
        myQueue.add("a");
        System.out.println("isFull:"+myQueue.isFull()+" isEmpty:"+myQueue.isEmpty());
        myQueue.add("b");
        myQueue.add("c");
        myQueue.add("d");
        myQueue.add("e");
        System.out.println("isFull:"+myQueue.isFull()+" isEmpty:"+myQueue.isEmpty());
        myQueue.add("f");
    }

我们定义长度是5的队列，分别加入a b c d e f6个元素，并且看一下空和满的状态。

打印日志如下：

isFull:false isEmpty:true
isFull:false isEmpty:false
isFull:true isEmpty:false
Exception in thread "main" java.lang.RuntimeException: 队列已经满了
    at org.example.queue.MyQueue.add(MyQueue.java:29)
    at org.example.queue.MyQueue.main(MyQueue.java:82)

空和满的状态都是对的，而且再插入f元素的时候，报错了”队列已经满了“，是没有问题的。出队的测试这里就不做了，留个小伙伴们去做吧。

并发与等待

队列的基础代码已经实现了，我们再看看有没有其他的问题。对了，第一个问题就是并发，我们多个线程同时入队或者出队时，就会引发问题，那么怎么办呢？其实也很简单，加上synchronized关键字就可以了，如下：

/**
 * 入队操作
 * @param e
 * @return
 */
public synchronized boolean add(T e) {
    if (isFull()) {
        throw new RuntimeException("队列已经满了");
    }
    data[tail] = e;
    tail = (tail + 1) % size;
    if (tail == front) {
        flag = 1;
    }

    return true;
}

/**
 * 出队操作
 * @return
 */
public synchronized T poll() {
    if (isEmpty()) {
        throw new RuntimeException("队列中没有元素");
    }
    T rtnData = data[front];
    front = (front + 1) % size;
    if (front == tail) {
        flag = 0;
    }
    return rtnData;
}

这样入队出队操作就不会有并发的问题了。下面我们再去解决上面小伙伴提出的问题，就是入队时，队列满了要等待，出队时，队列空了要等待，这个要怎么解决呢？这里要用的wait()和notifyAll()了，再进行编码前，我们先理清一下思路，

目前队列的长度是5，并且已经满了；
现在要向队列插入第6个元素，插入时，判断队列满了，要进行等待wait();
此时有一个出队操作，队列有空位了，此时应该唤起之前等待的线程，插入元素；

相反的，出队时，队列是空的，也要等待，当队列有元素时，唤起等待的线程，进行出队操作。好了，撸代码，

/**
 * 入队操作
 * @param e
 * @return
 */
public synchronized boolean add(T e) throws InterruptedException {
    if (isFull()) {
        wait();
    }
    data[tail] = e;
    tail = (tail + 1) % size;
    if (tail == front) {
        flag = 1;
    }
    notifyAll();
    return true;
}

/**
 * 出队操作
 * @return
 */
public synchronized T poll() throws InterruptedException {
    if (isEmpty()) {
        wait();
    }
    T rtnData = data[front];
    front = (front + 1) % size;
    if (front == tail) {
        flag = 0;
    }
    notifyAll();
    return rtnData;
}

之前我们抛异常的地方，统一改成了wait()，而且方法执行到最后进行notifyAll()，唤起等待的线程。我们进行简单的测试，

public static void main(String[] args) throws InterruptedException {
    MyQueue<String> myQueue = new MyQueue<>(5);
    new Thread(() -> {
        try {
            System.out.println(myQueue.poll());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }).start();

    myQueue.add("a");
}

测试结果没有问题，可以正常打印"a"。这里只进行了出队的等待测试，入队的测试，小伙伴们自己完成吧。

if还是while

到这里，我们手撸的消息队列还算不错，基本的功能都实现了，但是有没有什么问题呢？我们看看下面的测试程序，

public static void main(String[] args) throws InterruptedException {
    MyQueue<String> myQueue = new MyQueue<>(5);
    new Thread(() -> {
        try {
            System.out.println(myQueue.poll());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }).start();
    new Thread(() -> {
        try {
            System.out.println(myQueue.poll());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }).start();

    Thread.sleep(5000);
    myQueue.add("a");
}

我们启动了两个消费者线程，同时从队列里获取数据，此时，队列是空的，两个线程都进行等待，5秒后，我们插入元素"a"，看看结果如何，

a
null

结果两个消费者都打印出了日志，一个获取到null，一个获取到”a“，这是什么原因呢？还记得我们怎么判断空和满的吗？对了，使用的是if，我们捋一下整体的过程，

两个消费者线程同时从队列获取数据，队列是空的，两个消费者通过if判断，进入等待；
5秒后，向队列中插入"a"元素，并唤起所有等待线程；
两个消费者线程被依次唤起，一个取到值，一个没有取到。没有取到是因为取到的线程将front加了1导致的。这里为什么说依次唤起等待线程呢？因为notifyAll()不是同时唤起所有等待线程，是依次唤起，而且顺序是不确定的。

我们希望得到的结果是，一个消费线程得到”a“元素，另一个消费线程继续等待。这个怎么实现呢？对了，就是将判断是用到的if改为while，如下：

/**
 * 入队操作
 * @param e
 * @return
 */
public synchronized boolean add(T e) throws InterruptedException {
    while (isFull()) {
        wait();
    }
    data[tail] = e;
    tail = (tail + 1) % size;
    if (tail == front) {
        flag = 1;
    }
    notifyAll();
    return true;
}

/**
 * 出队操作
 * @return
 */
public synchronized T poll() throws InterruptedException {
    while (isEmpty()) {
        wait();
    }
    T rtnData = data[front];
    front = (front + 1) % size;
    if (front == tail) {
        flag = 0;
    }
    notifyAll();
    return rtnData;
}

在判断空还是满的时候，我们使用while去判断，当两个消费线程被依次唤起时，还会再进行空和满的判断，这时，第一个消费线程判断队列中有元素，会进行获取，第二个消费线程被唤起时，判断队列没有元素，会再次进入等待。我们写段代码测试一下，

public static void main(String[] args) throws InterruptedException {
    MyQueue<String> myQueue = new MyQueue<>(5);
    new Thread(() -> {
        try {
            System.out.println(myQueue.poll());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }).start();
    new Thread(() -> {
        try {
            System.out.println(myQueue.poll());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }).start();

    Thread.sleep(5000);
    myQueue.add("a");
    Thread.sleep(5000);
    myQueue.add("b");
}

同样，有两个消费线程去队列获取数据，此时队列为空，然后，我们每隔5秒，插入一个元素，看看结果如何，