字節(jié)二面：了解環(huán)形隊(duì)列嗎？有哪些使用場景？

作者：朱晉君 2025-01-15 07:54:02

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

大家好，我是君哥。

在日常開發(fā)工作中，環(huán)形隊(duì)列的使用并不多，但其實(shí)環(huán)形隊(duì)列是一個(gè)很有用的數(shù)據(jù)結(jié)構(gòu)，而且有不少使用場景。今天來聊一聊環(huán)形隊(duì)列的使用場景。

1.環(huán)形隊(duì)列

隊(duì)列這個(gè)數(shù)據(jù)結(jié)構(gòu)最大的特點(diǎn)就是先進(jìn)先出，它可以有兩種實(shí)現(xiàn)方式，無界隊(duì)列一般用鏈表來實(shí)現(xiàn)，有界隊(duì)列可以用數(shù)組來實(shí)現(xiàn)。

但使用數(shù)組來實(shí)現(xiàn)隊(duì)列，有新數(shù)據(jù)插入時(shí)，需要搬移元素，造成額外的性能開銷。要解決數(shù)據(jù)搬移的問題，我們可以考慮使用環(huán)形隊(duì)列。

下圖是一個(gè) 8 個(gè)元素的環(huán)形隊(duì)列：

圖片

環(huán)形隊(duì)列的特點(diǎn)是寫完最后一個(gè)元素后接著從頭開始寫，讀元素也一樣。初始狀態(tài)，head 和 tail 都指向數(shù)據(jù)下標(biāo)是 0 的位置。每寫入一個(gè)元素，tail 往后移動(dòng)一個(gè)指針，每讀取一個(gè)元素，head 指針往后移動(dòng)一個(gè)指針。如果寫入的速度超過了讀取速度一圈，未讀取的元素就會(huì)被覆蓋。

下圖是用數(shù)組來表示的環(huán)形隊(duì)列，尾節(jié)點(diǎn)指向頭結(jié)點(diǎn)，實(shí)現(xiàn)首尾相連：

圖片

在上圖中，head 所在數(shù)組位置元素值是 3，tail 所在數(shù)組位置元素值是 7。這時(shí)如果我們插入一個(gè)新元素 a，環(huán)形隊(duì)列變成下圖：

圖片

那環(huán)形隊(duì)列代碼怎么實(shí)現(xiàn)呢？這里給出一個(gè)示例代碼：

public class CircleQueue {
 //實(shí)現(xiàn)環(huán)形隊(duì)列的數(shù)組
 private String[] items;
 //數(shù)組大小
 private int size;
 //數(shù)組元素?cái)?shù)量
 private int count = 0;
 private int head = 0;
 private int tail = 0;
 //申請(qǐng)一個(gè)指定容量的隊(duì)列
 public CircleQueue(int size){
  items = new String[size];
  this.size = size;
 }

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

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

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

2.使用場景

2.1 延時(shí)消息

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

在 RocketMQ 5.0 之前，4.x 版本定義了 18 個(gè)延時(shí)級(jí)別：

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

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

這樣的設(shè)計(jì)存在一個(gè)問題，延時(shí)級(jí)別只有 18 個(gè)，不太靈活，對(duì)于大型的復(fù)雜業(yè)務(wù)系統(tǒng)，延時(shí)級(jí)別可能成千上萬，這種設(shè)計(jì)無法滿足。

為了解決這個(gè)設(shè)計(jì)問題，RocketMQ 5.0 基于時(shí)間輪算法引入了定時(shí)消息。如下圖：

圖片

圖中定義了一個(gè) 60s 的時(shí)間輪，時(shí)間輪上有一個(gè)指向當(dāng)前時(shí)間的指針定時(shí)地移動(dòng)到下一秒時(shí)間。

這樣不用去輪詢所有消息，每一個(gè)時(shí)間節(jié)點(diǎn)上的消息用鏈表串起來，當(dāng)時(shí)間輪上的指針移動(dòng)到當(dāng)前的時(shí)間時(shí)，這個(gè)時(shí)間節(jié)點(diǎn)上符合條件的消息就交給異步線程來處理。

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

2.2 Disruptor

Disruptor 是一款高性能的消息隊(duì)列，它使用到了環(huán)形隊(duì)列這個(gè)數(shù)據(jù)結(jié)構(gòu)。那 Disruptor 使用環(huán)形隊(duì)列是怎樣做到高性能的呢？

2.2.1 內(nèi)存預(yù)分配

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

2.2.2 無鎖并發(fā)

Disruptor 是一種生產(chǎn)者-消費(fèi)者模式，當(dāng)多個(gè)生產(chǎn)者在同一個(gè)位置寫事件消息時(shí)，就會(huì)被覆蓋。如下圖，線程1 把位置 1 的元素更新成 b，線程 2 寫入時(shí)本來應(yīng)該在位置 2 寫入 c，但是寫入了位置 1，導(dǎo)致覆蓋了線程 1 寫入的值。消費(fèi)者并發(fā)消費(fèi)時(shí)也有類似的問題。

圖片

解決這個(gè)問題最好的方法就是給寫入的代碼加鎖，只允許獲取到鎖的線程執(zhí)行，但這樣失去了并發(fā)優(yōu)勢(shì)，性能降低。

為了解決加鎖帶來的性能問題，Disruptor 在設(shè)計(jì)上進(jìn)行了改造。當(dāng)一個(gè)線程要寫入循環(huán)隊(duì)列時(shí)，先申請(qǐng)隊(duì)列上連續(xù)的 n 個(gè)位置，申請(qǐng)成功這 n 個(gè)位置是線程獨(dú)享的，這樣線程在寫入元素時(shí)就不用擔(dān)心被覆蓋。消費(fèi)者進(jìn)行并發(fā)消費(fèi)時(shí)，也是先申請(qǐng)連續(xù)的 n 個(gè)位置獨(dú)自消費(fèi)，跟其他線程互相隔離。