Sentinel源码—6.熔断降级和数据统计的实现

告陕无 · 4 天前

大纲
1.DegradeSlot实现熔断降级的原理与源码
2.Sentinel数据指标统计的滑动窗口算法

1.DegradeSlot实现熔断降级的原理与源码
(1)熔断降级规则DegradeRule的配置Demo
(2)注册熔断降级监听器和加载熔断降级规则
(3)DegradeSlot根据熔断降级规则对请求进行验证

(1)熔断降级规则DegradeRule的配置Demo
首先熔断降级规则的应用场景有如下两种：

场景一：在微服务架构中，当一个服务出现问题时，可以通过配置熔断降级规则，防止故障扩散，保护整个系统的稳定性。

场景二：在调用第三方API时，可以配置熔断降级规则，避免因第三方API不稳定导致自身系统不稳定。

然后从下图可知，熔断降级规则包含以下属性：

属性一：熔断策略(grade)
这表示的是熔断降级规则的类型，取值范围分别是：

RuleConstant.DEGRADE_GRADE_RT(慢调用比例)
RuleConstant.DEGRADE_GRADE_EXCEPTION_RATIO(异常比例)
RuleConstant.DEGRADE_GRADE_EXCEPTION_COUNT(异常数)

复制代码

其中，默认下的熔断降级规则是基于慢调用比例策略的，也就是默认值为：

RuleConstant.DEGRADE_GRADE_RT

复制代码

属性二：熔断降级的阈值(count)
DegradeRule.count属性的具体含义取决于DegradeRule.grade属性的值。

如果grade为慢调用比例，则count表示慢调用比例阈值。
如果grade为异常比例，则count表示异常比例阈值。
如果grade为异常数，则count表示异常数阈值。

复制代码

属性三：熔断时长(timeWindow)
这表示的是熔断降级发生后的降级持续时间，在这段时间内对应的资源将被降级。

属性四：最小请求数(minRequestAmount)
这表示的是熔断降级统计周期内的最小请求总数。仅当周期内的请求总数达到此值时，才会根据grade和count进行熔断降级。默认值为：

RuleConstant.DEGRADE_DEFAULT_MIN_REQUEST_AMOUNT

复制代码

属性五：慢调用比例阈值(slowRatioThreshold)
该属性当grade为慢调用比例时生效，取值范围为0到1之间的小数，表示慢调用请求占总请求的比例。

属性六：统计时长(statIntervalMs)
这表示的是熔断降级统计周期(单位：毫秒)，默认值为1000毫秒(1秒)。在这个周期内，Sentinel会对请求进行统计，以判断是否要进行熔断降级。

public class DegradeRule extends AbstractRule {
//熔断策略，表示的是熔断降级规则的类型
private int grade = RuleConstant.DEGRADE_GRADE_RT;
//熔断降级的阈值，具体含义取决于DegradeRule.grade属性的值
//如果grade为慢调用比例，则count表示慢调用比例阈值
//如果grade为异常比例，则count表示异常比例阈值
//如果grade为异常数，则count表示异常数阈值
private double count;
//熔断时长，即熔断降级发生后的降级持续时间，在这段时间内对应的资源将被降级
private int timeWindow;
//最小请求数，仅当周期内的请求总数达到此值时，才会根据grade和count进行熔断降级
private int minRequestAmount = RuleConstant.DEGRADE_DEFAULT_MIN_REQUEST_AMOUNT;
//慢调用比例阈值，仅当grade为慢调用比例时生效
private double slowRatioThreshold = 1.0d;
//统计时长，熔断降级统计周期，在这个周期内，Sentinel会对请求进行统计，以判断是否要进行熔断降级
private int statIntervalMs = 1000;
...
}

复制代码

接着如下便是DegradeRule的配置Demo：

//Run this demo, and the output will be like:
//1529399827825,total:0, pass:0, block:0
//1529399828825,total:4263, pass:100, block:4164
//1529399829825,total:19179, pass:4, block:19176 // circuit breaker opens
//1529399830824,total:19806, pass:0, block:19806
//1529399831825,total:19198, pass:0, block:19198
//1529399832824,total:19481, pass:0, block:19481
//1529399833826,total:19241, pass:0, block:19241
//1529399834826,total:17276, pass:0, block:17276
//1529399835826,total:18722, pass:0, block:18722
//1529399836826,total:19490, pass:0, block:19492
//1529399837828,total:19355, pass:0, block:19355
//1529399838827,total:11388, pass:0, block:11388
//1529399839829,total:14494, pass:104, block:14390 // After 10 seconds, the system restored
//1529399840854,total:18505, pass:0, block:18505
//1529399841854,total:19673, pass:0, block:19676
public class SlowRatioCircuitBreakerDemo {
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 {
initDegradeRule();
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);
pass.incrementAndGet();
//RT: [40ms, 60ms)
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();
}
}
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
.setCount(50)
.setTimeWindow(10)//Retry timeout (in second)
.setSlowRatioThreshold(0.6)//Circuit breaker opens when slow request ratio > 60%
.setMinRequestAmount(100)
.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(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);
}
}
}

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册