sentinel 簡介

概述

Sentinel 是面向分布式、多語言異構(gòu)化服務(wù)架構(gòu)的流量治理組件，主要以流量為切入點(diǎn)，從流量路由、流量控制、流量整形、熔斷降級、系統(tǒng)自適應(yīng)過載保護(hù)、熱點(diǎn)流量防護(hù)等多個(gè)維度來幫助開發(fā)者保障微服務(wù)的穩(wěn)定性。

支持功能：

1. 流量控制
2. 熔斷降級
3. 系統(tǒng)自適應(yīng)保護(hù)
4. 集群流量控制
5. 網(wǎng)關(guān)流量控制
6. 熱點(diǎn)參數(shù)限流
7. 來源訪問控制

基于sentinel實(shí)現(xiàn)的功能

1. 流量控制和限流 基于可視化控制臺
2. 實(shí)現(xiàn)熔斷降級

1.實(shí)現(xiàn)流量控制

基于 QPS/并發(fā)數(shù)的流量控制
基于調(diào)用關(guān)系的流量控制

1.1 啟動控制臺服務(wù)及主要概述說明

1.1.1 配置

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
</dependency>

spring:
  cloud:
    sentinel:
      transport:
        dashboard: localhost:8080

1.1.2 啟動控制臺

首先下載所需版本的sentinel可視化服務(wù)jar包，下載地址https://github.com/alibaba/Sentinel/releases
默認(rèn)參數(shù)：
端口：8080
用戶名：sentinel
密碼：sentinel

1.1.3 基于可視化控制臺設(shè)置QPS或并發(fā)線程數(shù)控制訪問

資源名：流控針對的方法名；
針對來源：需要在代碼中指定 在方法中添加ContextUtil.enter(resourceName,"填寫來源名稱") 下面代碼中有示例；
閾值類型：QPS、并發(fā)線程數(shù)；
1). QPS：每秒請求查詢數(shù)量
2). 并發(fā)線程數(shù)：瞬時(shí)的請求數(shù)量
單機(jī)閾值：限制的訪問次數(shù)
支持集群,可設(shè)置集群單機(jī)訪問也可設(shè)置集群總訪問量
流控模式：直接、關(guān)聯(lián)、鏈路；
1). 關(guān)聯(lián)資源：如果流控模式為關(guān)聯(lián)時(shí)，才有該參數(shù)；
2). 入口資源：如果流控模式為鏈路時(shí)，才有該參數(shù)；
流控效果：快速失敗、Warm Up、排隊(duì)等待；
1). 預(yù)熱時(shí)長：如果流控效果為Warm Up才有該參數(shù)

流控模式：
直接：代表請求請求到該接口如果達(dá)到設(shè)置到閾值將直接進(jìn)行流控，直接拋出異常
關(guān)聯(lián)：代表請求到該接口，關(guān)聯(lián)的資源達(dá)到設(shè)置的閾值，然后才觸發(fā)流控，比如在同一個(gè)服務(wù)中會有兩個(gè)接口，一個(gè)讀資源操作的接口，一個(gè)是寫資源操作的接口，如果讀資源操作的接口QPS太大時(shí)，會影響寫操作業(yè)務(wù)完整性，即可以在設(shè)定資源名為寫操作的流控規(guī)則時(shí)流控模式設(shè)定為關(guān)聯(lián)模式，關(guān)聯(lián)的資源為讀操作的接口，就是當(dāng)讀操作的QPS達(dá)到閾值時(shí)，將寫操作進(jìn)行流控。（當(dāng)關(guān)聯(lián)當(dāng)資源達(dá)到閾值時(shí)，就限流自己）
鏈路：比如一個(gè)微服務(wù)中的兩個(gè)接口都調(diào)用了該微服務(wù)中的同一個(gè)service方法，并且該方法用SentinelResource注解給標(biāo)注了，然后對該注解標(biāo)注的資源進(jìn)行規(guī)則配置，則可以選擇鏈路模式，規(guī)則中資源名，就是注解標(biāo)注的名稱，入口資源就是指定，對那個(gè)接口調(diào)用該service的方法進(jìn)行流控。

流控效果:
快速失敗：顧名思義，就是直接拋出異常
Warm Up ：可以設(shè)置預(yù)熱的時(shí)長，即預(yù)熱/冷啟動方式。當(dāng)系統(tǒng)長期處于低水位的情況下，當(dāng)流量突然增加時(shí)，直接把系統(tǒng)拉升到高水位可能瞬間把系統(tǒng)壓垮。通過"冷啟動"，讓通過的流量緩慢增加，在一定時(shí)間內(nèi)逐漸增加到閾值上限，給冷系統(tǒng)一個(gè)預(yù)熱的時(shí)間，避免冷系統(tǒng)被壓垮。
排隊(duì)等待：顧名思義，就是請求一個(gè)一個(gè)排隊(duì)等待，設(shè)定一個(gè)請求等待的超時(shí)時(shí)間，等待時(shí)間超過設(shè)定的值，然后拋出異常。

官網(wǎng)地址：https://github.com/alibaba/Sentinel/wiki/%E6%B5%81%E9%87%8F%E6%8E%A7%E5%88%B6

1.1.4 可以通過注解形式控制方法的訪問

通過在方法上添加@SentinelResource注解對該方法進(jìn)行流控，如果超過流控規(guī)則限制，則返回下面方法的錯誤提示
注：兩個(gè)方法的返回值要一致；
value值要和方法名一致，blockHandler值要和下面方法名一致；
兩個(gè)方法的參數(shù)一定要一致（除BlockException外）；
下面的方法一定要有BlockException異常的參數(shù)

    @GetMapping(value = "/get")
    @SentinelResource(value = "getMethod",blockHandler = "fail")
    public String getMethod(){
        String resourceName = "get"; //一般即為請求路徑
        ContextUtil.enter(resourceName,"填寫來源名稱")
        return "SUCCESS";
    }

    public String fail(BlockException e){
        return "已被限制訪問";
    }

如果不使用可視化控制臺添加流控規(guī)則，而是在系統(tǒng)中自定義開發(fā)流控規(guī)則配置功能，則需要通過spring AOP功能來完成sentinel源碼改造工作相關(guān)重要屬性如下表
具體改造工作可以參考sentinel原理https://blog.csdn.net/wanger5354/article/details/122493842

2.實(shí)現(xiàn)熔斷降級

2.1 降級策略

2.1.1 降級策略(sentinel 1.8.0及以上版本)：

慢調(diào)用比例：需要設(shè)置慢調(diào)用的RT(即最大相應(yīng)時(shí)間，單位毫秒)，請求的相應(yīng)時(shí)間大于該值，則統(tǒng)計(jì)為慢調(diào)用。
當(dāng)單位統(tǒng)計(jì)時(shí)長內(nèi)，請求數(shù)目大于設(shè)置的最小請求數(shù)目并且慢調(diào)用的比例大于閾值，則接下來的熔斷時(shí)長內(nèi)請求會自動被熔斷。
經(jīng)過熔斷時(shí)長后熔斷器會進(jìn)入探測恢復(fù)狀態(tài)（HALF-OPEN 狀態(tài)），若接下來的一個(gè)請求響應(yīng)時(shí)間小于設(shè)置的慢調(diào)用RT,則結(jié)束熔斷，若大于設(shè)置的慢調(diào)用RT,則會再次被熔斷。

異常比例：當(dāng)單位統(tǒng)計(jì)時(shí)長內(nèi)請求數(shù)目大于設(shè)置的最小請求數(shù)目，并且異常的比例大于閾值，則接下來的熔斷時(shí)長內(nèi)請求會自動被熔斷。
經(jīng)過熔斷時(shí)長后熔斷器會進(jìn)入探測恢復(fù)狀態(tài)（HALF-OPEN 狀態(tài)），若接下來的一個(gè)請求成功完成（沒有錯誤）則結(jié)束熔斷，否則會再次被熔斷。
異常比率的閾值范圍是 [0.0, 1.0]，代表 0% - 100%。

異常數(shù)：當(dāng)單位統(tǒng)計(jì)時(shí)長內(nèi)的異常數(shù)目超過閾值之后會自動進(jìn)行熔斷。
經(jīng)過熔斷時(shí)長后熔斷器會進(jìn)入探測恢復(fù)狀態(tài)（HALF-OPEN 狀態(tài)），若接下來的一個(gè)請求成功完成（沒有錯誤）則結(jié)束熔斷，否則會再次被熔斷。

參數(shù)說明：
資源名：需要做熔斷處理的接口名
熔斷策略：慢比例調(diào)用、異常比例、異常數(shù)
最大RT(毫秒)：設(shè)置接口響應(yīng)時(shí)間比較標(biāo)準(zhǔn)值，如果接口響應(yīng)時(shí)間超過最大RT，則判定該次接口調(diào)用為慢調(diào)用
比例閾值(取值[0.0~1.0])：設(shè)置 慢調(diào)用的次數(shù)占有的比例或異常次數(shù)占有的比例，如果超過設(shè)置的比例閾值，則滿足熔斷條件的50%
熔斷時(shí)長：如果滿足熔斷條件，則熔斷的時(shí)長
最小請求數(shù)：設(shè)置單位時(shí)間內(nèi)接口請求次數(shù)標(biāo)準(zhǔn)值，如果超過設(shè)置的最小請求數(shù)，則滿足熔斷條件的50%
統(tǒng)計(jì)時(shí)長(毫秒)：單位統(tǒng)計(jì)時(shí)長，設(shè)置多長時(shí)間作為一個(gè)單位統(tǒng)計(jì)時(shí)長，默認(rèn)1000ms
異常數(shù)：設(shè)置單位時(shí)間內(nèi)，調(diào)用接口異常次數(shù)比較標(biāo)準(zhǔn)值

注意異常降級僅針對業(yè)務(wù)異常，對 Sentinel 限流降級本身的異常（BlockException）不生效。為了統(tǒng)計(jì)異常比例或異常數(shù)，需要通過 Tracer.trace(ex) 記錄業(yè)務(wù)異常。 示例：

Entry entry = null;
try {
    entry = SphU.entry(key, EntryType.IN, key);

    // Write your biz code here.
    // <<BIZ CODE>>
} catch (Throwable t) {
    if (!BlockException.isBlockException(t)) {
        Tracer.trace(t);
    }
} finally {
    if (entry != null) {
        entry.exit();
    }
}

2.1.2 降級策略(sentinel 1.8.0以下版本)：

RT(平均響應(yīng)時(shí)間，秒級)：平均響應(yīng)時(shí)間超出閾值且在時(shí)間窗口內(nèi)通過的請求>=5,兩個(gè)條件同時(shí)滿足后出發(fā)降級，窗口期過后關(guān)閉斷路器；RT最大4900 (更大需要通過-Dcsp.sentinel.statistic.max.rt=XXX才能生效)
異常比例(秒級)：QPS>=5且異常比例(秒級統(tǒng)計(jì))超過閾值時(shí)，觸發(fā)降級；時(shí)間窗口結(jié)束后，關(guān)閉降級
異常數(shù)(分鐘級)：異常數(shù)超過閾值時(shí)，觸發(fā)降級；時(shí)間窗口結(jié)束后，關(guān)閉降級
對比Hystrix熔斷降級比較，Hystrix是有半開狀態(tài)的，服務(wù)自動去檢測是否請求有異常，沒有異常就關(guān)閉斷路器恢復(fù)使用，有異常則繼續(xù)打開斷路器不可使用

2.2 重要參數(shù)

如果通過apollo等配置中心進(jìn)行配置時(shí)，重要參數(shù)非常重要；

2.3 熔斷器事件監(jiān)聽

Sentinel 支持注冊自定義的事件監(jiān)聽器監(jiān)聽熔斷器狀態(tài)變換事件（state change event）。示例:

EventObserverRegistry.getInstance().addStateChangeObserver("logging",
    (prevState, newState, rule, snapshotValue) -> {
        if (newState == State.OPEN) {
            // 變換至 OPEN state 時(shí)會攜帶觸發(fā)時(shí)的值
            System.err.println(String.format("%s -> OPEN at %d, snapshotValue=%.2f", prevState.name(),
                TimeUtil.currentTimeMillis(), snapshotValue));
        } else {
            System.err.println(String.format("%s -> %s at %d", prevState.name(), newState.name(),
                TimeUtil.currentTimeMillis()));
        }
    });

監(jiān)聽demo類，官網(wǎng)都提供demo，也可以直接去官網(wǎng)找，這里我只是搬運(yùn)工，嘻嘻?。?！
慢調(diào)用比例 監(jiān)聽示例SlowRatioDemo.java類

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.csp.sentinel.slots.block.degrade.circuitbreaker.CircuitBreaker.State;
import com.alibaba.csp.sentinel.slots.block.degrade.circuitbreaker.CircuitBreakerStrategy;
import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRule;
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRuleManager;
import com.alibaba.csp.sentinel.slots.block.degrade.circuitbreaker.EventObserverRegistry;
import com.alibaba.csp.sentinel.util.TimeUtil;

/**
 * 慢調(diào)用比例 監(jiān)聽示例
 */
public class SlowRatioDemo {

    private static final String KEY = "some_method";

    private static volatile boolean stop = false;
    private static int seconds = 120;

    private static AtomicInteger total = new AtomicInteger();
    private static AtomicInteger pass = new AtomicInteger();
    private static AtomicInteger block = new AtomicInteger();

    public static void main(String[] args) throws Exception {
        // 初始化加載熔斷規(guī)則
        initDegradeRule();
        // 熔斷器事件監(jiān)聽
        registerStateChangeObserver();
        startTick();

        int concurrency = 8;
        for (int i = 0; i < concurrency; i++) {
            Thread entryThread = new Thread(() -> {
                while (true) {
                    Entry entry = null;
                    try {
                        entry = SphU.entry(KEY);
//                        get();
                        pass.incrementAndGet();
                        // RT: [40ms, 60ms)  su
                        sleep(ThreadLocalRandom.current().nextInt(40, 60));
                    } catch (BlockException e) {
                        block.incrementAndGet();
                        sleep(ThreadLocalRandom.current().nextInt(5, 10));
                    } finally {
                        total.incrementAndGet();
                        if (entry != null) {
                            entry.exit();
                        }
                    }
                }
            });
            entryThread.setName("sentinel-simulate-traffic-task-" + i);
            entryThread.start();
        }
    }

    // 熔斷器事件監(jiān)聽
    private static void registerStateChangeObserver() {
        EventObserverRegistry.getInstance().addStateChangeObserver("logging",
                (prevState, newState, rule, snapshotValue) -> {
                    if (newState == State.OPEN) {
                        System.err.println(String.format("%s -> OPEN at %d, snapshotValue=%.2f", prevState.name(),
                                TimeUtil.currentTimeMillis(), snapshotValue));
                    } else {
                        System.err.println(String.format("%s -> %s at %d", prevState.name(), newState.name(),
                                TimeUtil.currentTimeMillis()));
                    }
                });
    }

    private static void initDegradeRule() {
        List<DegradeRule> rules = new ArrayList<>();
        DegradeRule rule = new DegradeRule(KEY)
                // 熔斷策略
                .setGrade(CircuitBreakerStrategy.SLOW_REQUEST_RATIO.getType())
                // Max allowed response time  RT 慢調(diào)用標(biāo)準(zhǔn)值 接口相應(yīng)時(shí)長超過RT則被判定為慢調(diào)用
                .setCount(50)
                // Retry timeout (in second)   窗口期  熔斷時(shí)長
                .setTimeWindow(10)
                // Circuit breaker opens when slow request ratio > 60%  慢調(diào)用比例   判定是否熔斷的條件之一
                .setSlowRatioThreshold(0.3)
                // 單位時(shí)長最小請求數(shù)   判定是否熔斷的條件之一
                .setMinRequestAmount(10)
                // 統(tǒng)計(jì)時(shí)長也叫單位時(shí)長
                .setStatIntervalMs(20000);
        rules.add(rule);

        DegradeRuleManager.loadRules(rules);
        System.out.println("Degrade rule loaded: " + rules);
    }

    private static void sleep(int timeMs) {
        try {
            TimeUnit.MILLISECONDS.sleep(timeMs);
        } catch (InterruptedException e) {
            // ignore
        }
    }

    private static void startTick() {
        Thread timer = new Thread(new TimerTask());
        timer.setName("sentinel-timer-tick-task");
        timer.start();
    }

    static class TimerTask implements Runnable {
        @Override
        public void run() {
            long start = System.currentTimeMillis();
            System.out.println("Begin to run! Go go go!");
            System.out.println("See corresponding metrics.log for accurate statistic data");

            long oldTotal = 0;
            long oldPass = 0;
            long oldBlock = 0;

            while (!stop) {
                sleep(1000);

                long globalTotal = total.get();
                long oneSecondTotal = globalTotal - oldTotal;
                oldTotal = globalTotal;

                long globalPass = pass.get();
                long oneSecondPass = globalPass - oldPass;
                oldPass = globalPass;

                long globalBlock = block.get();
                long oneSecondBlock = globalBlock - oldBlock;
                oldBlock = globalBlock;

                System.out.println(Thread.currentThread().getName()+"\t\t"+TimeUtil.currentTimeMillis() + ", total:" + oneSecondTotal
                        + ", pass:" + oneSecondPass + ", block:" + oneSecondBlock);

                if (seconds-- <= 0) {
                    stop = true;
                }
            }

            long cost = System.currentTimeMillis() - start;
            System.out.println("time cost: " + cost + " ms");
            System.out.println("total: " + total.get() + ", pass:" + pass.get()
                    + ", block:" + block.get());
            System.exit(0);
        }
    }


    public static String get() {
        System.out.println("===========SUCCESS==============");
        // RT: [40ms, 60ms)
        sleep(ThreadLocalRandom.current().nextInt(40, 60));
        return "Success";
    }

}

系統(tǒng)自適應(yīng)保護(hù) demoSystemGuardDemo.java類

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.alibaba.csp.sentinel.util.TimeUtil;
import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.EntryType;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.csp.sentinel.slots.system.SystemRule;
import com.alibaba.csp.sentinel.slots.system.SystemRuleManager;

/**
 * 系統(tǒng)自適應(yīng)保護(hù) demo
 *
 * 系統(tǒng)保護(hù)規(guī)則是從應(yīng)用級別的入口流量進(jìn)行控制，從單臺機(jī)器的總體 Load、RT、入口 QPS 和線程數(shù)四個(gè)維度監(jiān)控應(yīng)用數(shù)據(jù)，讓系統(tǒng)盡可能跑在最大吞吐量的同時(shí)保證系統(tǒng)整體的穩(wěn)定性。
 *
 * 系統(tǒng)保護(hù)規(guī)則是應(yīng)用整體維度的，而不是資源維度的，并且僅對入口流量生效。入口流量指的是進(jìn)入應(yīng)用的流量（EntryType.IN），比如 Web 服務(wù)或 Dubbo 服務(wù)端接收的請求，都屬于入口流量。
 *
 * 系統(tǒng)規(guī)則支持以下的閾值類型：
 *
 * Load（僅對 Linux/Unix-like 機(jī)器生效）：當(dāng)系統(tǒng) load1 超過閾值，且系統(tǒng)當(dāng)前的并發(fā)線程數(shù)超過系統(tǒng)容量時(shí)才會觸發(fā)系統(tǒng)保護(hù)。系統(tǒng)容量由系統(tǒng)的 maxQps * minRt 計(jì)算得出。設(shè)定參考值一般是 CPU cores * 2.5。
 * CPU usage（1.5.0+ 版本）：當(dāng)系統(tǒng) CPU 使用率超過閾值即觸發(fā)系統(tǒng)保護(hù)（取值范圍 0.0-1.0）。
 * RT：當(dāng)單臺機(jī)器上所有入口流量的平均 RT 達(dá)到閾值即觸發(fā)系統(tǒng)保護(hù)，單位是毫秒。
 * 線程數(shù)：當(dāng)單臺機(jī)器上所有入口流量的并發(fā)線程數(shù)達(dá)到閾值即觸發(fā)系統(tǒng)保護(hù)。
 * 入口 QPS：當(dāng)單臺機(jī)器上所有入口流量的 QPS 達(dá)到閾值即觸發(fā)系統(tǒng)保護(hù)。
 *
 */
public class SystemGuardDemo {

    private static AtomicInteger pass = new AtomicInteger();
    private static AtomicInteger block = new AtomicInteger();
    private static AtomicInteger total = new AtomicInteger();

    private static volatile boolean stop = false;
    private static final int threadCount = 100;

    private static int seconds = 60 + 40;

    public static void main(String[] args) throws Exception {

        tick();
        initSystemRule();

        for (int i = 0; i < threadCount; i++) {
            Thread entryThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        Entry entry = null;
                        try {
                            entry = SphU.entry("methodA", EntryType.IN);
                            pass.incrementAndGet();
                            try {
                                TimeUnit.MILLISECONDS.sleep(20);
                            } catch (InterruptedException e) {
                                // ignore
                            }
                        } catch (BlockException e1) {
                            block.incrementAndGet();
                            try {
                                TimeUnit.MILLISECONDS.sleep(20);
                            } catch (InterruptedException e) {
                                // ignore
                            }
                        } catch (Exception e2) {
                            // biz exception
                        } finally {
                            total.incrementAndGet();
                            if (entry != null) {
                                entry.exit();
                            }
                        }
                    }
                }

            });
            entryThread.setName("working-thread");
            entryThread.start();
        }
    }

    private static void initSystemRule() {
        List<SystemRule> rules = new ArrayList<SystemRule>();
        SystemRule rule = new SystemRule();
        // max load is 3
        rule.setHighestSystemLoad(3.0);
        // max cpu usage is 60%
        rule.setHighestCpuUsage(0.6);
        // max avg rt of all request is 10 ms
        rule.setAvgRt(10);
        // max total qps is 20
        rule.setQps(20);
        // max parallel working thread is 10
        rule.setMaxThread(10);

        rules.add(rule);
        SystemRuleManager.loadRules(Collections.singletonList(rule));
    }

    private static void tick() {
        Thread timer = new Thread(new TimerTask());
        timer.setName("sentinel-timer-task");
        timer.start();
    }

    static class TimerTask implements Runnable {
        @Override
        public void run() {
            System.out.println("begin to statistic!!!");
            long oldTotal = 0;
            long oldPass = 0;
            long oldBlock = 0;
            while (!stop) {
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                }
                long globalTotal = total.get();
                long oneSecondTotal = globalTotal - oldTotal;
                oldTotal = globalTotal;

                long globalPass = pass.get();
                long oneSecondPass = globalPass - oldPass;
                oldPass = globalPass;

                long globalBlock = block.get();
                long oneSecondBlock = globalBlock - oldBlock;
                oldBlock = globalBlock;

                System.out.println(seconds + ", " + TimeUtil.currentTimeMillis() + ", total:"
                        + oneSecondTotal + ", pass:"
                        + oneSecondPass + ", block:" + oneSecondBlock);
                if (seconds-- <= 0) {
                    stop = true;
                }
            }
            System.exit(0);
        }
    }
}