Sentinel原始碼解析系列:
1.Sentinel原始碼分析—FlowRuleManager載入規則做了什麼?

2. Sentinel原始碼分析—Sentinel是如何進行流量統計的?

3. Sentinel原始碼分析— QPS流量控制是如何實現的?

4.Sentinel原始碼分析— Sentinel是如何做到降級的?


這篇文章主要學習一下Sentinel如何實現自適應限流的。
為什麼要做自適應限流,官方給了兩個理由:

  1. 保證系統不被拖垮
  2. 在系統穩定的前提下,保持系統的吞吐量

我再貼一下官方的原理:

  1. 能夠保證水管裡的水量,能夠讓水順暢的流動,則不會增加排隊的請求;也就是說,這個時候的系統負載不會進一步惡化。
  2. 當保持入口的流量是水管出來的流量的最大的值的時候,可以最大利用水管的處理能力。
    更加具體的原理解釋可以看官方:系統自適應限流

所以看起來好像很厲害的樣子,所以我們來看看具體實現吧。

例子:

  1. 設定系統自適應規則
List<SystemRule> rules = new ArrayList<SystemRule>();
SystemRule rule = new SystemRule();
//限制最大負載
rule.setHighestSystemLoad(3.0);
// cpu負載60%
rule.setHighestCpuUsage(0.6);
// 設定平均響應時間 10 ms
rule.setAvgRt(10);
// 設定qps is 20
rule.setQps(20);
// 設定最大執行緒數 10
rule.setMaxThread(10);

rules.add(rule);
SystemRuleManager.loadRules(Collections.singletonList(rule));
  1. 設定限流
Entry entry = null;
try {
    entry = SphU.entry("methodA", EntryType.IN); 
    //dosomething
} catch (BlockException e1) {
    block.incrementAndGet();
    //dosomething
} catch (Exception e2) {
    // biz exception
} finally { 
    if (entry != null) {
        entry.exit();
    }
}

注意:系統保護規則是應用整體維度的,而不是資源維度的,並且僅對入口流量生效。入口流量指的是進入應用的流量(EntryType.IN),比如 Web 服務或 Dubbo 服務端接收的請求,都屬於入口流量。

我們先講一下SystemRuleManager這個類在初始化的時候做了什麼吧。

SystemRuleManager

private static SystemStatusListener statusListener = null;
@SuppressWarnings("PMD.ThreadPoolCreationRule")
private final static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1,
    new NamedThreadFactory("sentinel-system-status-record-task", true));

static {
    checkSystemStatus.set(false);
    statusListener = new SystemStatusListener();
    scheduler.scheduleAtFixedRate(statusListener, 5, 1, TimeUnit.SECONDS);
    currentProperty.addListener(listener);
}

SystemRuleManager初始化的時候會呼叫靜態程式碼塊,然後用scheduler執行緒池定時呼叫SystemStatusListener類的run方法。我們進入到SystemStatusListener類裡看一下:

SystemStatusListener#run

public void run() {
    try {
        OperatingSystemMXBean osBean = ManagementFactory.getPlatformMXBean(OperatingSystemMXBean.class);
        currentLoad = osBean.getSystemLoadAverage();
       
        currentCpuUsage = osBean.getSystemCpuLoad();

        StringBuilder sb = new StringBuilder();
        if (currentLoad > SystemRuleManager.getHighestSystemLoad()) {
            sb.append("load:").append(currentLoad).append(";");
            sb.append("cpu:").append(currentCpuUsage).append(";");
            sb.append("qps:").append(Constants.ENTRY_NODE.passQps()).append(";");
            sb.append("rt:").append(Constants.ENTRY_NODE.avgRt()).append(";");
            sb.append("thread:").append(Constants.ENTRY_NODE.curThreadNum()).append(";");
            sb.append("success:").append(Constants.ENTRY_NODE.successQps()).append(";");
            sb.append("minRt:").append(Constants.ENTRY_NODE.minRt()).append(";");
            sb.append("maxSuccess:").append(Constants.ENTRY_NODE.maxSuccessQps()).append(";");
            RecordLog.info(sb.toString());
        }

    } catch (Throwable e) {
        RecordLog.info("could not get system error ", e);
    }
}

這個方法用來做兩件事:

  1. 定時收集全域性資源情況,並列印日誌
  2. 給全域性變數currentLoad和currentCpuUsage賦值,用來做限流使用。

然後看一下SystemRuleManager.loadRules方法。SystemRuleManager和其他的規則管理是一樣的,當呼叫loadRules方法的時候會呼叫內部的listener並觸發它的configUpdate方法。
在SystemRuleManager中實現類了一個SystemPropertyListener,最終SystemRuleManager.loadRules方法會呼叫到SystemPropertyListener的configUpdate中。

SystemPropertyListener#configUpdate

public void configUpdate(List<SystemRule> rules) {
    restoreSetting();
    // systemRules = rules;
    if (rules != null && rules.size() >= 1) {
        for (SystemRule rule : rules) {
            loadSystemConf(rule);
        }
    } else {
        checkSystemStatus.set(false);
    }

    RecordLog.info(String.format("[SystemRuleManager] Current system check status: %s, "
            + "highestSystemLoad: %e, "
            + "highestCpuUsage: %e, "
            + "maxRt: %d, "
            + "maxThread: %d, "
            + "maxQps: %e",
        checkSystemStatus.get(),
        highestSystemLoad,
        highestCpuUsage,
        maxRt,
        maxThread,
        qps));
}

這個方法很簡單,首先是呼叫restoreSetting,用來重置rule的屬性,然後遍歷rule呼叫loadSystemConf對規則進行設定:

SystemRuleManager#loadSystemConf

public static void loadSystemConf(SystemRule rule) {
    boolean checkStatus = false;
    // Check if it's valid.

    if (rule.getHighestSystemLoad() >= 0) {
        highestSystemLoad = Math.min(highestSystemLoad, rule.getHighestSystemLoad());
        highestSystemLoadIsSet = true;
        checkStatus = true;
    }

    if (rule.getHighestCpuUsage() >= 0) {
        highestCpuUsage = Math.min(highestCpuUsage, rule.getHighestCpuUsage());
        highestCpuUsageIsSet = true;
        checkStatus = true;
    }

    if (rule.getAvgRt() >= 0) {
        maxRt = Math.min(maxRt, rule.getAvgRt());
        maxRtIsSet = true;
        checkStatus = true;
    }
    if (rule.getMaxThread() >= 0) {
        maxThread = Math.min(maxThread, rule.getMaxThread());
        maxThreadIsSet = true;
        checkStatus = true;
    }

    if (rule.getQps() >= 0) {
        qps = Math.min(qps, rule.getQps());
        qpsIsSet = true;
        checkStatus = true;
    }

    checkSystemStatus.set(checkStatus);

}

這些屬性都是在限流控制中會用到的屬性,無論設定哪個屬性都會設定checkStatus=true表示開啟系統自適應限流。

在設定好限流規則後會進入到SphU.entry方法中,通過建立slot鏈呼叫到SystemSlot,這裡是系統自適應限流的地方。

SystemSlot#entry

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                  boolean prioritized, Object... args) throws Throwable {
      //檢查一下是否符合限流條件,符合則進行限流
    SystemRuleManager.checkSystem(resourceWrapper);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

SystemRuleManager#checkSystem

public static void checkSystem(ResourceWrapper resourceWrapper) throws BlockException {
    // Ensure the checking switch is on.
    if (!checkSystemStatus.get()) {
        return;
    }
    //如果不是入口流量,那麼直接返回
    // for inbound traffic only
    if (resourceWrapper.getType() != EntryType.IN) {
        return;
    }

    // total qps
    double currentQps = Constants.ENTRY_NODE == null ? 0.0 : Constants.ENTRY_NODE.successQps();
    if (currentQps > qps) {
        throw new SystemBlockException(resourceWrapper.getName(), "qps");
    }

    // total thread
    int currentThread = Constants.ENTRY_NODE == null ? 0 : Constants.ENTRY_NODE.curThreadNum();
    if (currentThread > maxThread) {
        throw new SystemBlockException(resourceWrapper.getName(), "thread");
    }

    double rt = Constants.ENTRY_NODE == null ? 0 : Constants.ENTRY_NODE.avgRt();
    if (rt > maxRt) {
        throw new SystemBlockException(resourceWrapper.getName(), "rt");
    }

    // load. BBR algorithm.
    if (highestSystemLoadIsSet && getCurrentSystemAvgLoad() > highestSystemLoad) {
        if (!checkBbr(currentThread)) {
            throw new SystemBlockException(resourceWrapper.getName(), "load");
        }
    }

    // cpu usage
    if (highestCpuUsageIsSet && getCurrentCpuUsage() > highestCpuUsage) {
        if (!checkBbr(currentThread)) {
            throw new SystemBlockException(resourceWrapper.getName(), "cpu");
        }
    }
}

這個方法首先會校驗一下checkSystemStatus狀態和EntryType是不是IN,如果不是則直接返回。
然後對Constants.ENTRY_NODE進行操作。這個物件是一個final static 修飾的變數,代表是全域性物件。

public final static ClusterNode ENTRY_NODE = new ClusterNode();

所以這裡的限流操作都是對全域性其作用的,而不是對資源起作用。ClusterNode還是繼承自StatisticNode,所以最後都是呼叫StatisticNode的successQps、curThreadNum、avgRt,這幾個方法我的前幾篇文章都已經講過了,感興趣的可以自己去翻一下,這裡就不過多涉及了。

在下面呼叫getCurrentSystemAvgLoad方法和getCurrentCpuUsage方法呼叫到SystemStatusListener設定的全域性變數currentLoad和currentCpuUsage。這兩個引數是SystemRuleManager的定時任務定時收集的,忘了的同學回到上面講解SystemRuleManager的地方看一下。

在做load判斷和cpu usage判斷的時候會還會呼叫checkBbr方法來判斷:

private static boolean checkBbr(int currentThread) {
    if (currentThread > 1 &&
        currentThread > Constants.ENTRY_NODE.maxSuccessQps() * Constants.ENTRY_NODE.minRt() / 1000) {
        return false;
    }
    return true;
}

也就是說:當系統 load1 超過閾值,且系統當前的併發執行緒數超過系統容量時才會觸發系統保護。系統容量由系統的 maxQps * minRt 計算得出。

StatisticNode#maxSuccessQps

public double maxSuccessQps() {
    return rollingCounterInSecond.maxSuccess() * rollingCounterInSecond.getSampleCount();
}

maxSuccessQps方法是用視窗內的最大成功呼叫數和視窗數量相乘rollingCounterInSecond的視窗1秒的視窗數量是2,最大成功呼叫數如下得出:
ArrayMetric#maxSuccess

public long maxSuccess() {
    data.currentWindow();
    long success = 0;

    List<MetricBucket> list = data.values();
    for (MetricBucket window : list) {
        if (window.success() > success) {
            success = window.success();
        }
    }
    return Math.max(success, 1);
}

最大成功呼叫數是通過整個遍歷整個視窗,獲取所有窗口裡面最大的呼叫數。所以這樣的最大的併發量是一個預估值,不是真實值。

看到這裡我們再來看一下Constants.ENTRY_NODE的資訊是怎麼被收集的。
我在分析StatisticSlot這個類的時候有一段程式碼我當時也沒看懂有什麼用,現在就迎刃而解了:
StatisticSlot#entry

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                  boolean prioritized, Object... args) throws Throwable {
    try { 
            ....
        if (resourceWrapper.getType() == EntryType.IN) {
            // Add count for global inbound entry node for global statistics.
            Constants.ENTRY_NODE.increaseThreadNum();
            Constants.ENTRY_NODE.addPassRequest(count);
        }
            ....
    } catch (PriorityWaitException ex) {
            ....
        if (resourceWrapper.getType() == EntryType.IN) {
            // Add count for global inbound entry node for global statistics.
            Constants.ENTRY_NODE.increaseThreadNum();
        }
         ....
    } catch (BlockException e) {
            ....
        if (resourceWrapper.getType() == EntryType.IN) {
            // Add count for global inbound entry node for global statistics.
            Constants.ENTRY_NODE.increaseBlockQps(count);
        }
            .... 
        throw e;
    } catch (Throwable e) {
         ....
        if (resourceWrapper.getType() == EntryType.IN) {
            Constants.ENTRY_NODE.increaseExceptionQps(count);
        }
        throw e;
    }
}

在StatisticSlot的entry方法裡有很多對於type的判斷,如果是EntryType.IN,那麼就呼叫Constants.ENTRY_NODE的靜態方法進行資料的收集。

所以看到這裡我們可以知道,在前面有很多看不懂的程式碼其實只要慢慢琢磨,打個標記,那麼在後面的解析的過程中還是能夠慢慢看懂的。