sentinel 滑動窗口統計機制

sentinel的滑動窗口統計機制就是根據當前時間，獲取對應的時間窗口，並更新該時間窗口中的各項統計指標（pass/block/rt等），這些指標被用來進行后續判斷，比如限流、降級等；隨着時間的推移，當前時間點對應的時間窗口是變化的，這時會涉及到時間窗口的初始化、復用等。可以說，sentinel上的功能所用到的數據幾乎都是滑動窗口統計機制來維護和更新的。

 

sentinel 處理流程是基於slot鏈(ProcessorSlotChain)來完成的，比如限流、熔斷等，其中重要的一個slot就是StatisticSlot，它是做各種數據統計的，而限流/熔斷的數據判斷來源就是StatisticSlot，StatisticSlot的各種數據統計都是基於滑動窗口來完成的，因此本文就重點分析StatisticSlot的滑動窗口統計機制。

 

sentinel 的slot鏈(ProcessorSlotChain)是責任鏈模式的體現，那SlotChain是在哪創建的呢？是在 CtSph.lookProcessChain()方法中創建的，並且該方法會根據當前請求的資源先去一個靜態的HashMap中獲取，如果獲取不到才會創建，創建后會保存到HashMap中。這就意味着，同一個資源會全局共享一個SlotChain。默認生成ProcessorSlotChain邏輯為：

// DefaultSlotChainBuilder
public ProcessorSlotChain build() {
   ProcessorSlotChain chain = new DefaultProcessorSlotChain();
   chain.addLast(new NodeSelectorSlot());
   chain.addLast(new ClusterBuilderSlot());
   chain.addLast(new LogSlot());
   chain.addLast(new StatisticSlot());
   chain.addLast(new SystemSlot());
   chain.addLast(new AuthoritySlot());
   chain.addLast(new FlowSlot());
   chain.addLast(new DegradeSlot());

   return chain;
}

 

整個處理過程從第一個slot往后一直傳遞到最后一個的，當到達StatisticSlot時，開始統計各項指標，統計的結果又會被后續的Slot所采用，作為各種規則校驗的依據。各種指標如下：

public enum MetricEvent {
   PASS, // Normal pass.
   BLOCK, // Normal block.
   EXCEPTION, // 異常統計
   SUCCESS,
   RT, // rt統計
   OCCUPIED_PASS
}

 

StatisticSlot.entry流程

處理流程走到StatisticSlot時，首先觸發后續slot.entry方法，然后統計各項指標，后續slot中數據判斷來源就是這里統計的各項指標。StatisticSlot.entry 邏輯如下：

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNodenode, int count, Object... args) throws Throwable {
   try {
       // 觸發下一個Slot的entry方法
       fireEntry(context, resourceWrapper, node, count, args);
       // 如果能通過SlotChain中后面的Slot的entry方法，說明沒有被限流或降級
       // 統計信息
       node.increaseThreadNum();
       node.addPassRequest();
       // 省略部分代碼
  } catch (BlockException e) {
       context.getCurEntry().setError(e);
       // Add block count.
       node.increaseBlockedQps();
       // 省略部分代碼
       throw e;
  } catch (Throwable e) {
       context.getCurEntry().setError(e);
       // Should not happen
       node.increaseExceptionQps();
       // 省略部分代碼
       throw e;
  }
}

由以上代碼可知，StatisticSlot主要就做了3件事：

1. 觸發后續slot的entry方法，進行規則校驗

2. 校驗通過則更新node實時指標數據

3. 校驗不通過則更新node異常指標數據

注意：由於后續的fireEntry操作和更新本次統計信息是兩個操作，不是原子的，會造成限流不准的小問題，比如設置的FlowRule count為20，並發情況下可能稍大於20，不過針對大部分場景來說，這點偏差是可以容忍的，畢竟我們要的是限流效果，而不是必須精確的限流操作。

 

更新node實時指標數據

我們可以看到 node.addPassRequest() 這段代碼是在fireEntry執行之后執行的，這意味着，當前請求通過了sentinel的流控等規則，此時需要將當次請求記錄下來，也就是執行 node.addPassRequest() 這行代碼，具體的代碼如下所示：

// DefaultNode
public void addPassRequest() {
   super.addPassRequest();
   this.clusterNode.addPassRequest();
}

這里的node是一個 DefaultNode 實例，這里特別補充一個 DefaultNode 和 ClusterNode 的區別：

• DefaultNode：保存着某個resource在某個context中的實時指標，每個DefaultNode都指向一個ClusterNode。

• ClusterNode：保存着某個resource在所有的context中實時指標的總和，同樣的resource會共享同一個ClusterNode，不管他在哪個context中。

 

上面代碼不管是 DefaultNode 還是 ClusterNode ，走的都是StatisticNode 對象的 addPassRequest 方法：

private transient volatile Metric rollingCounterInSecond = new ArrayMetric(2, 1000);
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000);

public void addPassRequest(int count) {
   rollingCounterInSecond.addPass(count); // 對每秒指標統計
   rollingCounterInMinute.addPass(count); // 每分鍾指標統計
}

 

每一個通過的指標（pass）都是調用Metric 的接口進行操作的，並且是通過 ArrayMetric 這種實現類，代碼如下：

public ArrayMetric(int windowLength, int interval) {
   this.data = new WindowLeapArray(windowLength, interval);
}

public void addPass(int count) {
   // 獲取當前時間窗口
   WindowWrap<MetricBucket> wrap = data.currentWindow();
   wrap.value().addPass(count);
}

 

首先通過 currentWindow() 獲取當前時間窗口，然后更新當前時間窗口對應的統計指標，以下代碼重點關注幾個判斷邏輯：

// LeapArray
public WindowWrap<T> currentWindow() {
   return currentWindow(TimeUtil.currentTimeMillis());
}
// TimeUtil
public static long currentTimeMillis() {
   // currentTimeMillis是由一個tick線程每個1ms更新一次，具體邏輯在TimeUtil類中
   return currentTimeMillis;
}
// LeapArray
public WindowWrap<T> currentWindow(long timeMillis) {
   // 計算當前時間點落在滑動窗口的下標
   int idx = calculateTimeIdx(timeMillis);
   // Calculate current bucket start time.
   long windowStart = calculateWindowStart(timeMillis);

   // 獲取當前時間點對應的windowWrap，array為AtomicReferenceArray
   while (true) {
       WindowWrap<T> old = array.get(idx);
       if (old == null) {
           // 1.為空表示當前時間窗口為初始化過，創建WindowWrap並cas設置到array中
           WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs,windowStart, newEmptyBucket());
           if (array.compareAndSet(idx, null, window)) {
               return window;
          } else {
               Thread.yield();
          }
      } else if (windowStart == old.windowStart()) {
           // 2.獲取的時間窗口正好對應當前時間，直接返回
           return old;
      } else if (windowStart > old.windowStart()) {
           // 3.獲取的時間窗口為老的，進行reset操作復用
           if (updateLock.tryLock()) {
               try {
                   return resetWindowTo(old, windowStart);
              } finally {
                   updateLock.unlock();
              }
          } else {
               Thread.yield();
          }
      } else if (windowStart < old.windowStart()) {
           // 4.時間回撥了，正常情況下不會走到這里
           return new WindowWrap<T>(windowLengthInMs, windowStart,newEmptyBucket());
      }
  }
}

 

獲取當前時間窗口對應的WindowWrap之后，就可以進行更新操作了。

// wrap.value().addPass(count);
public void addPass(int n) {
   add(MetricEvent.PASS, n);
}
// MetricBucket
public MetricBucket add(MetricEvent event, long n) {
   // 對應MetricEvent枚舉中值
   counters[event.ordinal()].add(n);
   return this;
}

到這里為止，整個指標統計流程就完成了，下面重點看下滑動窗口機制。

 

滑動窗口機制

時間窗口是用WindowWrap對象表示的，其屬性如下：

private final long windowLengthInMs;  // 時間窗口的長度
private long windowStart; // 時間窗口開始時間
private T value; // MetricBucket對象，保存各個指標數據

 

sentinel時間基准由tick線程來做，每1ms更新一次時間基准，邏輯如下：

currentTimeMillis = System.currentTimeMillis();
Thread daemon = new Thread(new Runnable() {
   @Override
   public void run() {
       while (true) {
           currentTimeMillis = System.currentTimeMillis();
           try {
               TimeUnit.MILLISECONDS.sleep(1);
          } catch (Throwable e) {
          }
      }
  }
});
daemon.setDaemon(true);
daemon.setName("sentinel-time-tick-thread");
daemon.start();

 

sentinel默認有每秒和每分鍾的滑動窗口，對應的LeapArray類型，它們的初始化邏輯是：

protected int windowLengthInMs; // 單個滑動窗口時間值
protected int sampleCount; // 滑動窗口個數
protected int intervalInMs; // 周期值（相當於所有滑動窗口時間值之和）

public LeapArray(int sampleCount, int intervalInMs) {
   this.windowLengthInMs = intervalInMs / sampleCount;
   this.intervalInMs = intervalInMs;
   this.sampleCount = sampleCount;

   this.array = new AtomicReferenceArray<WindowWrap<T>>(sampleCount);
}

針對每秒滑動窗口，windowLengthInMs=500，sampleCount=2，intervalInMs=1000，針對每分鍾滑動窗口，windowLengthInMs=1000，sampleCount=60，intervalInMs=60000，對應代碼：

private transient volatile Metric rollingCounterInSecond = new ArrayMetric(2, 1000);
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000);

currentTimeMillis時間基准（tick線程）每1ms更新一次，通過currentWindow(timeMillis)方法獲取當前時間點對應的WindowWrap對象，然后更新對應的各種指標，用於做限流、降級時使用。注意，當前時間基准對應的事件窗口初始化時lazy模式，並且會復用的。

 

Sentinel 底層采用高性能的滑動窗口數據結構 LeapArray 來統計實時的秒級指標數據，可以很好地支撐寫多於讀的高並發場景。最后以一張圖結束吧：

 

 

 往期精選 

• sentinel 核心概念

• 你的Redis有類轉換異常么

• Redis 基礎數據結構

• 別再問我ConcurrentHashMap了

• 分布式鎖設計與實現

• ConcurrentHashMap竟然也有死循環問題？

• 你的ThreadLocal線程安全么

• MySQL基礎概念知多少

覺得文章不錯，對你有所啟發和幫助，希望能轉發給更多的小伙伴。如果有問題，請關注下面公眾號，發送問題給我，多謝。
歡迎小伙伴 關注【TopCoder】閱讀更多精彩好文。