大家好，我是君哥。

在日常開發工作中，環形隊列的使用并不多，但其實環形隊列是一個很有用的數據結構，而且有不少使用場景。今天來聊一聊環形隊列的使用場景。

1.環形隊列

隊列這個數據結構最大的特點就是先進先出，它可以有兩種實現方式，無界隊列一般用鏈表來實現，有界隊列可以用數組來實現。

但使用數組來實現隊列，有新數據插入時，需要搬移元素，造成額外的性能開銷。要解決數據搬移的問題，我們可以考慮使用環形隊列。

下圖是一個 8 個元素的環形隊列：

圖片

環形隊列的特點是寫完最后一個元素后接著從頭開始寫，讀元素也一樣。初始狀態，head 和 tail 都指向數據下標是 0 的位置。每寫入一個元素，tail 往后移動一個指針，每讀取一個元素，head 指針往后移動一個指針。如果寫入的速度超過了讀取速度一圈，未讀取的元素就會被覆蓋。

下圖是用數組來表示的環形隊列，尾節點指向頭結點，實現首尾相連：

圖片

在上圖中，head 所在數組位置元素值是 3，tail 所在數組位置元素值是 7。這時如果我們插入一個新元素 a，環形隊列變成下圖：

圖片

那環形隊列代碼怎么實現呢？這里給出一個示例代碼：

public class CircleQueue {
 //實現環形隊列的數組
 private String[] items;
 //數組大小
 private int size;
 //數組元素數量
 private int count = 0;
 private int head = 0;
 private int tail = 0;
 //申請一個指定容量的隊列
 public CircleQueue(int size){
  items = new String[size];
  this.size = size;
 }

 public boolean enqueue(String item){
  if ((tail + 1) % size == head){
   //隊列滿
   return false;
  }
  items[tail] = item;
  tail = (tail + 1) % size;
  count++;
  return true;
 }

 public String dequeue(){
     String item = null;
  //隊列空
  if(head == tail){
   return item;
  }
  item = items[head];
  head = (head + 1) % size;
  count--;
  return item;
 }
}

在上面的例子中，如果隊列滿了，就會寫入消息失敗。不過在實際使用場景中，有些場景如果隊列滿了，可以覆蓋掉當前 tail 位置上的元素，tail 繼續往下一個位置移動。這個適用于丟失數據影響較小的場景，比如記錄日志。

2.使用場景

2.1 延時消息

在消息隊列中，延時消息的使用場景很多，比如超過 30 分鐘關閉未支付訂單。主流消息隊列實現延時關閉訂單的方式是采用線程輪詢的方式來判斷訂單是否超過 30 分鐘，如果超過則關閉訂單。

在 RocketMQ 5.0 之前，4.x 版本定義了 18 個延時級別：

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

Broker 收到消息后，會根據延時級別把消息保存到同一個 Topic（SCHEDULE_TOPIC_XXXX）下的不同 queue。然后啟動 18 個線程來對每個 queue 做輪詢判斷，如果時間到了，就把消息投遞到原始隊列，等待 Consumer 來拉取。

這樣的設計存在一個問題，延時級別只有 18 個，不太靈活，對于大型的復雜業務系統，延時級別可能成千上萬，這種設計無法滿足。

為了解決這個設計問題，RocketMQ 5.0 基于時間輪算法引入了定時消息。如下圖：

圖片

圖中定義了一個 60s 的時間輪，時間輪上有一個指向當前時間的指針定時地移動到下一秒時間。

這樣不用去輪詢所有消息，每一個時間節點上的消息用鏈表串起來，當時間輪上的指針移動到當前的時間時，這個時間節點上符合條件的消息就交給異步線程來處理。

如果一個消息的延時時間超過 60s，比如 130s，該怎么設置呢？在每個時間輪節點增加一個 round 字段，記錄時間輪轉動的圈數，對于延時 130s 的消息，round 就是 2，放在第 10 個時間刻度的鏈表中。時間輪每轉動一圈，round 值減一，這樣當時間輪轉到一個節點，處理節點上的消息時，首先判斷 round 值是否等于 0，如果等于 0，則把這個消息從鏈表中移出交給異步線程執行，否則將 round 減 1 繼續檢查后面的任務。

2.2 Disruptor

Disruptor 是一款高性能的消息隊列，它使用到了環形隊列這個數據結構。那 Disruptor 使用環形隊列是怎樣做到高性能的呢？

2.2.1 內存預分配

Disruptor 使用循環隊列，在隊列初始化的時候，數組元素一次性初始化，這樣可以不僅提升緩存命中率，還可以避免頻繁 GC。

2.2.2 無鎖并發

Disruptor 是一種生產者-消費者模式，當多個生產者在同一個位置寫事件消息時，就會被覆蓋。如下圖，線程1 把位置 1 的元素更新成 b，線程 2 寫入時本來應該在位置 2 寫入 c，但是寫入了位置 1，導致覆蓋了線程 1 寫入的值。消費者并發消費時也有類似的問題。

圖片

解決這個問題最好的方法就是給寫入的代碼加鎖，只允許獲取到鎖的線程執行，但這樣失去了并發優勢，性能降低。

為了解決加鎖帶來的性能問題，Disruptor 在設計上進行了改造。當一個線程要寫入循環隊列時，先申請隊列上連續的 n 個位置，申請成功這 n 個位置是線程獨享的，這樣線程在寫入元素時就不用擔心被覆蓋。消費者進行并發消費時，也是先申請連續的 n 個位置獨自消費，跟其他線程互相隔離。

2.2.3 解決偽共享

環形隊列內部數組使用緩存行填充技術來避免偽共享問題，進一步提高了性能。

2.3 日志收集

dmesg 這個 Linux 命令我們應該了解過，主要用于查看系統啟動時的日志信息、硬件信息。

dmesg 使用的日志就是存儲在環形緩存區中，每當有新的日志寫入時，如果環形隊列已滿，就會覆蓋舊的日志，這樣可以保證內核日志不會占用過多的內存空間，而且還能夠不斷記錄新日志。

3.總結

環形隊列作為一種有界循環隊列，在消息中間件、高性能內存隊列 Disruptor、日志收集等方面有廣泛的應用。了解循環隊列的原理，可以更好的理解它的使用場景。