聊聊sentinel的DegradeSlot

1
2
3
4
5
6
7
8
9
10
11
12
public class DegradeSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, Object... args)
throws Throwable {
DegradeRuleManager.checkDegrade(resourceWrapper, context, node, count);
fireEntry(context, resourceWrapper, node, count, args);
}
@Override
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
fireExit(context, resourceWrapper, count, args);
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
public class DegradeRuleManager {
private static volatile Map<String, List<DegradeRule>> degradeRules
= new ConcurrentHashMap<String, List<DegradeRule>>();
final static RulePropertyListener listener = new RulePropertyListener();
private static SentinelProperty<List<DegradeRule>> currentProperty
= new DynamicSentinelProperty<List<DegradeRule>>();
static {
currentProperty.addListener(listener);
}
/**
* Listen to the {@link SentinelProperty} for {@link DegradeRule}s. The property is the source
* of {@link DegradeRule}s. Degrade rules can also be set by {@link #loadRules(List)} directly.
*
* @param property the property to listen.
*/
public static void register2Property(SentinelProperty<List<DegradeRule>> property) {
synchronized (listener) {
currentProperty.removeListener(listener);
property.addListener(listener);
currentProperty = property;
}
}
public static void checkDegrade(ResourceWrapper resource, Context context, DefaultNode node, int count)
throws BlockException {
if (degradeRules == null) {
return;
}
List<DegradeRule> rules = degradeRules.get(resource.getName());
if (rules == null) {
return;
}
for (DegradeRule rule : rules) {
if (!rule.passCheck(context, node, count)) {
throw new DegradeException(rule.getLimitApp());
}
}
}
//......
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
/**
* <p>
* Degrade is used when the resources are in an unstable state, these resources
* will be degraded within the next defined time window. There are two ways to
* measure whether a resource is stable or not:
* </p>
* <ul>
* <li>
* Average response time ({@code DEGRADE_GRADE_RT}): When
* the average RT exceeds the threshold ('count' in 'DegradeRule', in milliseconds), the
* resource enters a quasi-degraded state. If the RT of next coming 5
* requests still exceed this threshold, this resource will be downgraded, which
* means that in the next time window (defined in 'timeWindow', in seconds) all the
* access to this resource will be blocked.
* </li>
* <li>
* Exception ratio: When the ratio of exception count per second and the
* success qps exceeds the threshold, access to the resource will be blocked in
* the coming window.
* </li>
* </ul>
*
* @author jialiang.linjl
*/
public class DegradeRule extends AbstractRule {
private static final int RT_MAX_EXCEED_N = 5;
private static ScheduledExecutorService pool = Executors.newScheduledThreadPool(
Runtime.getRuntime().availableProcessors(), new NamedThreadFactory("sentinel-degrade-reset-task", true));
/**
* RT threshold or exception ratio threshold count.
*/
private double count;
/**
* Degrade recover timeout (in seconds) when degradation occurs.
*/
private int timeWindow;
/**
* Degrade strategy (0: average RT, 1: exception ratio).
*/
private int grade = RuleConstant.DEGRADE_GRADE_RT;
private volatile boolean cut = false;
public int getGrade() {
return grade;
}
public void setGrade(int grade) {
this.grade = grade;
}
private AtomicLong passCount = new AtomicLong(0);
private final Object lock = new Object();
//......
@Override
public boolean passCheck(Context context, DefaultNode node, int acquireCount, Object... args) {
if (cut) {
return false;
}
ClusterNode clusterNode = ClusterBuilderSlot.getClusterNode(this.getResource());
if (clusterNode == null) {
return true;
}
if (grade == RuleConstant.DEGRADE_GRADE_RT) {
double rt = clusterNode.avgRt();
if (rt < this.count) {
passCount.set(0);
return true;
}
// Sentinel will degrade the service only if count exceeds.
if (passCount.incrementAndGet() < RT_MAX_EXCEED_N) {
return true;
}
} else {
double exception = clusterNode.exceptionQps();
double success = clusterNode.successQps();
long total = clusterNode.totalQps();
// if total qps less than RT_MAX_EXCEED_N, pass.
if (total < RT_MAX_EXCEED_N) {
return true;
}
if (success == 0) {
return exception < RT_MAX_EXCEED_N;
}
if (exception / (success + exception) < count) {
return true;
}
}
synchronized (lock) {
if (!cut) {
// Automatically degrade.
cut = true;
ResetTask resetTask = new ResetTask(this);
pool.schedule(resetTask, timeWindow, TimeUnit.SECONDS);
}
return false;
}
}
//......
private static final class ResetTask implements Runnable {
private DegradeRule rule;
ResetTask(DegradeRule rule) {
this.rule = rule;
}
@Override
public void run() {
rule.getPassCount().set(0);
rule.setCut(false);
}
}
}