sentinel的几个实体分析
sentinel中有几个比较重要的实体需要我们了解。
-
resource
resource是sentinel中最重要的一个概念。sentinel通过resource来保护具体的业务代码和其他后方服务。sentinel把复杂的逻辑给屏蔽掉了，用户只需要为受保护的代码或服务定义一个resource，然后定义rule就可以了，剩下的就通通交给sentinel来处理。并且resource和rule是解耦的，rule设置可以在runtime动态修改，定义完resource后，就可以通过在程序中埋点来保护自己的服务了，埋点的方式有两种：
- try-catch 方式（通过 SphU.entry(…)），当 catch 到BlockException时执行异常处理(或fallback)
- if-else 方式（通过 SphO.entry(…)），当返回 false 时执行异常处理(或fallback)
以上这两种方式都是通过硬编码的形式定义resource然后进行resource埋点的，对业务代码的侵入太大，从0.1.1版本开始， Sentinel 加入了注解的支持，可以通过注解来定义资源，具体的注解为：SentinelResource 。通过注解除了可以定义资源外，还可以指定 blockHandler 和 fallback 方法。
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
//这个注解代表定义一个sentinel resource
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface SentinelResource {
//sentinel resource名称
String value() default "";
//entry类型 （输入或者输出，默认是输出）
EntryType entryType() default EntryType.OUT;
//资源的分类(类型)
int resourceType() default 0;
//限流的异常名，默认为空
String blockHandler() default "";
/*
blockHandler默认和原方法在同一个类中。
如果希望使用其他类的函数，则需要指定 blockHandlerClass 为对应的类的 Class 对象。必须是静态的，否则无法解析
*/
Class<?>[] blockHandlerClass() default {};
//熔断方法名
String fallback() default "";
/*
defaultFallBack()用来做默认的降级方法
这个方法不接受任何参数，并且返回必须的类型和原始方法适配
*/
String defaultFallback() default "";
/*
fallback()默认情况下是和原方法在一起的。
同blockHandlerClass。
*/
Class<?>[] fallbackClass() default {};
//用来追踪的异常列表
Class<? extends Throwable>[] exceptionsToTrace() default {Throwable.class};
/*
指明有哪些异常可以被忽略。
exceptionsToTrace()和exceptionsToIgnore()不能够同时存在
exceptionsToIgnore()优先使用。
*/
Class<? extends Throwable>[] exceptionsToIgnore() default {};
}
其中EntryType的枚举为：
1
2
3
4
public enum EntryType {
IN("IN"),
OUT("OUT");
}
有了注解以后，接下来就是看看通过反射，获取配置了注解的方法是如何使用的了。使用了@SentinelResource就需要依赖sentinel-annotation-aspectj了。具体实现如下：
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
//SentinelResource切面的实现。继承于AbstractSentinelAspectSupport
@Aspect //定义切面
public class SentinelResourceAspect extends AbstractSentinelAspectSupport {
//匹配有SentinelResource注解的方法。设置切入点
@Pointcut("@annotation(com.alibaba.csp.sentinel.annotation.SentinelResource)")
public void sentinelResourceAnnotationPointcut() {
}
//以sentinelResourceAnnotationPointcut()为切入点，环绕增强。也就是说有这个注解的方法环绕增强
@Around("sentinelResourceAnnotationPointcut()")
public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable {
//反射，获取方法信息
Method originMethod = resolveMethod(pjp);
//通过方法反射出注解信息
SentinelResource annotation = originMethod.getAnnotation(SentinelResource.class);
//如果注解信息为null的话，就抛出异常。否则，继续
if (annotation == null) {
throw new IllegalStateException("Wrong state for SentinelResource annotation");
}
//获取资源名
String resourceName = getResourceName(annotation.value(), originMethod);
//获取entry类型
EntryType entryType = annotation.entryType();
//获取资源类型
int resourceType = annotation.resourceType();
Entry entry = null;
try {
//出发entry()出发熔断降级策略，具体看我的[sentinel原理](https://www.naffan.cn/tech/2020/04/04/01.html)
entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());
return pjp.proceed();
} catch (BlockException ex) {
//如果遇到了BlockException说明限流阻塞了。具体会去调用注解中设置的blockHandler()方法
return handleBlockException(pjp, annotation, ex);
} catch (Throwable ex) {
//反射注解中的忽略异常列表
Class<? extends Throwable>[] exceptionsToIgnore = annotation.exceptionsToIgnore();
// 如果异常忽略列表数量不为空且抛出的异常属于忽略异常，继续抛出异常。
if (exceptionsToIgnore.length > 0 && exceptionBelongsTo(ex, exceptionsToIgnore)) {
throw ex;
}
//如果抛出的异常属于需要跟踪的异常
if (exceptionBelongsTo(ex, annotation.exceptionsToTrace())) {
//跟踪，并且出发熔断降级方法，handleFallBack()
traceException(ex);
return handleFallback(pjp, annotation, ex);
}
// No fallback function can handle the exception, so throw it out.
throw ex;
} finally {
//最终不要忘了entry需要exit。回收资源用的
if (entry != null) {
entry.exit(1, pjp.getArgs());
}
}
}
}
我们看完了反射的注解了。但是其中有一些方法是继承于AbstractSentinelAspectSupport的实现方法。我们也来看看刚才用到的方法：
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
//一些给切面提供功能的普通方法
public abstract class AbstractSentinelAspectSupport {
//追踪记录异常，这些异常不是阻塞的异常
protected void traceException(Throwable ex) {
Tracer.trace(ex);
}
//判断ex和annotation反射出来的类型。然后进行追踪
protected void traceException(Throwable ex, SentinelResource annotation) {
Class<? extends Throwable>[] exceptionsToIgnore = annotation.exceptionsToIgnore();
// The ignore list will be checked first.
if (exceptionsToIgnore.length > 0 && exceptionBelongsTo(ex, exceptionsToIgnore)) {
return;
}
if (exceptionBelongsTo(ex, annotation.exceptionsToTrace())) {
traceException(ex);
}
}
//判断异常是否在提供的异常列表中
protected boolean exceptionBelongsTo(Throwable ex, Class<? extends Throwable>[] exceptions) {
if (exceptions == null) {
return false;
}
for (Class<? extends Throwable> exceptionClass : exceptions) {
if (exceptionClass.isAssignableFrom(ex.getClass())) {
return true;
}
}
return false;
}
//获取资源名称
protected String getResourceName(String resourceName, /*@NonNull*/ Method method) {
// If resource name is present in annotation, use this value.
if (StringUtil.isNotBlank(resourceName)) {
return resourceName;
}
// 调用方法类中的esolveMethodName。这个方法用因为涉及线程安全问题，所以加锁来获取资源名称。
return MethodUtil.resolveMethodName(method);
}
//调用熔断方法
protected Object handleFallback(ProceedingJoinPoint pjp, SentinelResource annotation, Throwable ex)
throws Throwable {
return handleFallback(pjp, annotation.fallback(), annotation.defaultFallback(), annotation.fallbackClass(), ex);
}
//真正执行熔断的方法
protected Object handleFallback(ProceedingJoinPoint pjp, String fallback, String defaultFallback,
Class<?>[] fallbackClass, Throwable ex) throws Throwable {
//获取参数
Object[] originArgs = pjp.getArgs();
// 如果配置了就执行熔断方法，没有配置就调用一个默认的熔断方法
Method fallbackMethod = extractFallbackMethod(pjp, fallback, fallbackClass);
if (fallbackMethod != null) {
// 构造参数
int paramCount = fallbackMethod.getParameterTypes().length;
Object[] args;
if (paramCount == originArgs.length) {
args = originArgs;
} else {
args = Arrays.copyOf(originArgs, originArgs.length + 1);
args[args.length - 1] = ex;
}
try {
//熔断方法是否是静态方法
if (isStatic(fallbackMethod)) {
//调用静态的熔断方法
return fallbackMethod.invoke(null, args);
}
//不是静态方法的话，就调用熔断方法
return fallbackMethod.invoke(pjp.getTarget(), args);
} catch (InvocationTargetException e) {
// throw the actual exception
throw e.getTargetException();
}
}
// If fallback is absent, we'll try the defaultFallback if provided.
return handleDefaultFallback(pjp, defaultFallback, fallbackClass, ex);
}
//默认的熔断方法。过程同上
protected Object handleDefaultFallback(ProceedingJoinPoint pjp, String defaultFallback,
Class<?>[] fallbackClass, Throwable ex) throws Throwable {
// Execute the default fallback function if configured.
Method fallbackMethod = extractDefaultFallbackMethod(pjp, defaultFallback, fallbackClass);
if (fallbackMethod != null) {
// Construct args.
Object[] args = fallbackMethod.getParameterTypes().length == 0 ? new Object[0] : new Object[] {ex};
try {
if (isStatic(fallbackMethod)) {
return fallbackMethod.invoke(null, args);
}
return fallbackMethod.invoke(pjp.getTarget(), args);
} catch (InvocationTargetException e) {
// throw the actual exception
throw e.getTargetException();
}
}
// If no any fallback is present, then directly throw the exception.
throw ex;
}
//调用阻塞限流，同上
protected Object handleBlockException(ProceedingJoinPoint pjp, SentinelResource annotation, BlockException ex)
throws Throwable {
// Execute block handler if configured.
Method blockHandlerMethod = extractBlockHandlerMethod(pjp, annotation.blockHandler(),
annotation.blockHandlerClass());
if (blockHandlerMethod != null) {
Object[] originArgs = pjp.getArgs();
// Construct args.
Object[] args = Arrays.copyOf(originArgs, originArgs.length + 1);
args[args.length - 1] = ex;
try {
if (isStatic(blockHandlerMethod)) {
return blockHandlerMethod.invoke(null, args);
}
return blockHandlerMethod.invoke(pjp.getTarget(), args);
} catch (InvocationTargetException e) {
// throw the actual exception
throw e.getTargetException();
}
}
// If no block handler is present, then go to fallback.
return handleFallback(pjp, annotation, ex);
}
//判断方法是否是静态方法
private boolean isStatic(Method method) {
return Modifier.isStatic(method.getModifiers());
}
…………
}
至此，通过@SentinelResource使用熔断降级，阻塞限流的原理就分析完成了。那接下来让我们看看ResourceWrapper是什么。
在 Sentinel 中具体表示资源的类是：ResourceWrapper ，他是一个抽象的包装类，包装了资源的 Name 和EntryType。他有两个实现类，分别是：StringResourceWrapper 和 MethodResourceWrapper
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
//resource的关于name和type的封装
public abstract class ResourceWrapper {
protected final String name;
//上面已经列出了EntryType的代码了，一个是IN，一个是OUT
protected final EntryType entryType;
protected final int resourceType;
//构造犯法，先判name和entryType是否是空或者null。如果不是的话就将其赋值给类内部变量。
public ResourceWrapper(String name, EntryType entryType, int resourceType) {
AssertUtil.notEmpty(name, "resource name cannot be empty");
AssertUtil.notNull(entryType, "entryType cannot be null");
this.name = name;
this.entryType = entryType;
this.resourceType = resourceType;
}
public String getName() {
return name;
}
public EntryType getEntryType() {
return entryType;
}
public int getResourceType() {
return resourceType;
}
//获取美化后的资源名称
public abstract String getShowName();
@Override
public int hashCode() {
return getName().hashCode();
}
@Override
public boolean equals(Object obj) {
if (obj instanceof ResourceWrapper) {
ResourceWrapper rw = (ResourceWrapper)obj;
return rw.getName().equals(getName());
}
return false;
}
}
从代码中，我们可以看到其实这就是一个简单的resource封装。在实际的使用过程中，作为传递的参数。这样的好处是传递时可以记录下当时的状态信息等。接下来我们看看两种继承类
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
//通过方法信息来封装resource
public class MethodResourceWrapper extends ResourceWrapper {
private final transient Method method;
public MethodResourceWrapper(Method method, EntryType e) {
//调用下面的方法
this(method, e, ResourceTypeConstants.COMMON);
}
public MethodResourceWrapper(Method method, EntryType e, int resType) {
//调用父类的构造方法。第一参数是资源名，这里通过方法类解析出方法名字符串格式
super(MethodUtil.resolveMethodName(method), e, resType);
this.method = method;
}
public Method getMethod() {
return method;
}
@Override
public String getShowName() {
return name;
}
//这个其实就是之后的resouce的Key
@Override
public String toString() {
return "MethodResourceWrapper{" +
"name='" + name + '\'' +
", entryType=" + entryType +
", resourceType=" + resourceType +
'}';
}
}
另一个是
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
//其实和上面的方法一个意思，只是下面的实现直接可以把resource的string类型取到了
public class StringResourceWrapper extends ResourceWrapper {
public StringResourceWrapper(String name, EntryType e) {
super(name, e, ResourceTypeConstants.COMMON);
}
public StringResourceWrapper(String name, EntryType e, int resType) {
super(name, e, resType);
}
@Override
public String getShowName() {
return name;
}
//这个名称和上面的有些许的不一样
@Override
public String toString() {
return "StringResourceWrapper{" +
"name='" + name + '\'' +
", entryType=" + entryType +
", resourceType=" + resourceType +
'}';
}
}
当然，这块可以自定义的。只要按照SPI的规则来即可。
-
slot
Slot 是另一个 Sentinel 中非常重要的概念， Sentinel 的工作流程就是围绕着一个个插槽所组成的插槽链来展开的。需要注意的是每个插槽都有自己的职责，他们各司其职完好的配合，通过一定的编排顺序，来达到最终的限流降级的目的。默认的各个插槽之间的顺序是固定的，因为有的插槽需要依赖其他的插槽计算出来的结果才能进行工作。
但是这并不意味着我们只能按照框架的定义来，Sentinel 通过 SlotChainBuilder 作为 SPI 接口，使得 Slot Chain 具备了扩展的能力。我们可以通过实现 SlotsChainBuilder 接口加入自定义的 slot 并自定义编排各个 slot 之间的顺序，从而可以给 Sentinel 添加自定义的功能。
那SlotChain是在哪创建的呢？是在 CtSph.lookProcessChain() 方法中创建的，并且该方法会根据当前请求的资源先去一个静态的HashMap中获取，如果获取不到才会创建，创建后会保存到HashMap中。这就意味着，同一个资源会全局共享一个SlotChain。
相关slot的更多讲解可以看我之前写的sentinel原理解析，上面介绍的slot都在里面讲到过
-
context
context是sentinel整个系统的上下文功能。context里面记载着很多关键信息。
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
/*
这个类保存着整个调用过程中的元数据，这其中包括
EntranceNode - 当前调用树的跟节点
Entry - 当前调用节点
Node - 和Entry有关的统计信息
origin - 先解释为源点吧。源点对我们区分控制发起者/消费者非常用用。一般源点可以视为消费者的服务app名称或者ip
每个SphU.entry或者SphO.entry应该都在一个context中，如果我们不明确的调用ContextUtil.enter()方法，那么会使用一个默认的context。
如果我们在同一个context中多次调用SphU.entry()一个调用树会被创建起来。
在不同的context中的相同的resource会分开来计算，详情请看NodeSelectorSlot的实现
*/
public class Context {
private final String name;
//当前调用树的入口node
private DefaultNode entranceNode;
//当前所在的entry
private Entry curEntry;
//当前context的源点信息 ，一般是消费者名称或者是个ip
private String origin = "";
private final boolean async;
//新建一个同步的上下文
public static Context newAsyncContext(DefaultNode entranceNode, String name) {
return new Context(name, entranceNode, true);
}
public Context(DefaultNode entranceNode, String name) {
this(name, entranceNode, false);
}
//建立context的最终方法。就是给context对象里设置这三个类内变量
public Context(String name, DefaultNode entranceNode, boolean async) {
this.name = name;
this.entranceNode = entranceNode;
this.async = async;
}
public boolean isAsync() {
return async;
}
public String getName() {
return name;
}
//下面是一堆get和set方法
public Node getCurNode() {
return curEntry == null ? null : curEntry.getCurNode();
}
public Context setCurNode(Node node) {
this.curEntry.setCurNode(node);
return this;
}
public Entry getCurEntry() {
return curEntry;
}
public Context setCurEntry(Entry curEntry) {
this.curEntry = curEntry;
return this;
}
public String getOrigin() {
return origin;
}
public Context setOrigin(String origin) {
this.origin = origin;
return this;
}
public double getOriginTotalQps() {
return getOriginNode() == null ? 0 : getOriginNode().totalQps();
}
public double getOriginBlockQps() {
return getOriginNode() == null ? 0 : getOriginNode().blockQps();
}
public double getOriginPassReqQps() {
return getOriginNode() == null ? 0 : getOriginNode().successQps();
}
public double getOriginPassQps() {
return getOriginNode() == null ? 0 : getOriginNode().passQps();
}
public long getOriginTotalRequest() {
return getOriginNode() == null ? 0 : getOriginNode().totalRequest();
}
public long getOriginBlockRequest() {
return getOriginNode() == null ? 0 : getOriginNode().blockRequest();
}
public double getOriginAvgRt() {
return getOriginNode() == null ? 0 : getOriginNode().avgRt();
}
public int getOriginCurThreadNum() {
return getOriginNode() == null ? 0 : getOriginNode().curThreadNum();
}
public DefaultNode getEntranceNode() {
return entranceNode;
}
//获取当前节点的父节点
public Node getLastNode() {
if (curEntry != null && curEntry.getLastNode() != null) {
return curEntry.getLastNode();
} else {
return entranceNode;
}
}
//获取源点
public Node getOriginNode() {
return curEntry == null ? null : curEntry.getOriginNode();
}
@Override
public String toString() {
return "Context{" +
"name='" + name + '\'' +
", entranceNode=" + entranceNode +
", curEntry=" + curEntry +
", origin='" + origin + '\'' +
", async=" + async +
'}';
}
}
可以看到，context中存储着entry和node等信息。等一会我会详细的分析这两个类。但是，我想想先看看context的创建实现。context的实现位置在ContextUtil.java中
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
//工具类，用来在当前线程中获取或者创建上下文
public class ContextUtil {
//存储在threadLocal中，为了方便存取
private static ThreadLocal<Context> contextHolder = new ThreadLocal<>();
//定义一个静态的且是volatile类型的hashMap来存储所有EntranceNode类，每个EntrnceNode和一个唯一的上下文名称绑定。 这个名称是Context.toString定义的
//为什么不用concurrnetHashMap呢？
private static volatile Map<String, DefaultNode> contextNameNodeMap = new HashMap<>();
//用一个可重入锁
private static final ReentrantLock LOCK = new ReentrantLock();
//nullContext这个也是一个类。一会看看他干了啥
private static final Context NULL_CONTEXT = new NullContext();
//静态代码块，直接运行
static {
//缓存EntranceNode
initDefaultContext();
}
private static void initDefaultContext() {
//默认的上下文名称
String defaultContextName = Constants.CONTEXT_DEFAULT_NAME;
//构造EntranceNode，采用字符串封装Resource的方式。EntryType为出口类型
EntranceNode node = new EntranceNode(new StringResourceWrapper(defaultContextName, EntryType.IN), null);
//将node添加在root下。
Constants.ROOT.addChild(node);
//存储在hashmap中
contextNameNodeMap.put(defaultContextName, node);
}
//测试用，非线程安全。重置hashMap用的
static void resetContextMap() {
if (contextNameNodeMap != null) {
RecordLog.warn("Context map cleared and reset to initial state");
contextNameNodeMap.clear();
initDefaultContext();
}
}
/*
进入调用上下文，它标记为调用链的入口。
context被ThreadLocal封起来了，意味着每个线程拥有着自己的context
如果当前线程没有context，就会创建一个context给这个线程。
一个context会与一个调用链中的一个代表着入口统计节点的EntranceNode进行绑定。
如果当前context没有EntranceNode，新的EntranceNode会被创建出来。
要注意的是同样的context名可能会被同样的EntraceNode全局共享。
在ClusterBuilderSlot类中会会创建源节点
不同的资源的不同源点会创建不同的Node，统计源点的总数等于resource名称量+源点数量
所以，当有很多源点时，应该仔细考虑内存占用率。
不同的context在同样的resource中会被单独计算，具体看NodeSelectorSlot
*/
public static Context enter(String name, String origin) {
//如果context name等于配置中的默认名称就抛出相应的exception
if (Constants.CONTEXT_DEFAULT_NAME.equals(name)) {
throw new ContextNameDefineException(
"The " + Constants.CONTEXT_DEFAULT_NAME + " can't be permit to defined!");
}
//调用下面的方法
return trueEnter(name, origin);
}
//真正实现创建context的实现。
protected static Context trueEnter(String name, String origin) {
//从threadLocal中获取context
Context context = contextHolder.get();
if (context == null) {
//如果context为空，首先定义一个空的hashMap
Map<String, DefaultNode> localCacheNameMap = contextNameNodeMap;
//从hashMap中找到key是name的value。
DefaultNode node = localCacheNameMap.get(name);
if (node == null) {
//node是空的话，hashMap的大小大于Constants.MAX_CONTEXT_NAME_SIZE
if (localCacheNameMap.size() > Constants.MAX_CONTEXT_NAME_SIZE) {
//把hasehMap设置为空
setNullContext();
return NULL_CONTEXT;
} else {
//node不空，需要上锁，用的重入锁
LOCK.lock();
try {
//double get
node = contextNameNodeMap.get(name);
if (node == null) {
//如果还是空的话，hashMap的大小大于Constants.MAX_CONTEXT_NAME_SIZE，跟上面的处理一样。
if (contextNameNodeMap.size() > Constants.MAX_CONTEXT_NAME_SIZE) {
setNullContext();
return NULL_CONTEXT;
} else {
//这块是重点。定义一个EntranceNode。
node = new EntranceNode(new StringResourceWrapper(name, EntryType.IN), null);
// 将Node添加到根下面，也就是构造调用树。
Constants.ROOT.addChild(node);
//再定义一个比上边hashMap空间大1的hashMap
Map<String, DefaultNode> newMap = new HashMap<>(contextNameNodeMap.size() + 1);
//将上边的hashMap放入新的HashMap中
newMap.putAll(contextNameNodeMap);
//然后，多出来的那一个空间用来存放当前的node
newMap.put(name, node);
//原先的HashMap更新为新的HashMap
contextNameNodeMap = newMap;
}
}
} finally {
//结束，解锁
LOCK.unlock();
}
}
}
//hashMap构造完毕以后，开始构造context。并保存到当前的ThreadLocal中。
context = new Context(node, name);
context.setOrigin(origin);
contextHolder.set(context);
}
//返回context，这个context
return context;
}
private static boolean shouldWarn = true;
//把hashMap设置为空。
private static void setNullContext() {
contextHolder.set(NULL_CONTEXT);
// Don't need to be thread-safe.
if (shouldWarn) {
RecordLog.warn("[SentinelStatusChecker] WARN: Amount of context exceeds the threshold "
+ Constants.MAX_CONTEXT_NAME_SIZE + ". Entries in new contexts will NOT take effect!");
shouldWarn = false;
}
}
//重载Context.enter()
public static Context enter(String name) {
return enter(name, "");
}
//从当前线程中退出context，也就是从ThreadLocal中删除Context
public static void exit() {
Context context = contextHolder.get();
if (context != null && context.getCurEntry() == null) {
contextHolder.set(null);
}
}
//获取当前entrance node hashMap的大小
public static int contextSize() {
return contextNameNodeMap.size();
}
//判断提供的context是不是初始自动创建额context
public static boolean isDefaultContext(Context context) {
if (context == null) {
return false;
}
return Constants.CONTEXT_DEFAULT_NAME.equals(context.getName());
}
//从当前的线程中获取context
public static Context getContext() {
return contextHolder.get();
}
//替换context的方法。
static Context replaceContext(Context newContext) {
Context backupContext = contextHolder.get();
if (newContext == null) {
contextHolder.remove();
} else {
contextHolder.set(newContext);
}
return backupContext;
}
//执行提供的context中的代码。可以看到这个f是个runnable接口实现。也就是说f是个可以进行扩展的方法。提供开发者自定义。
public static void runOnContext(Context context, Runnable f) {
Context curContext = replaceContext(context);
try {
f.run();
} finally {
replaceContext(curContext);
}
}
}
ContextUtil看完了，再看看NullContext。
1
2
3
4
5
6
7
8
//这个类比较简单，就是调用Context中的context方法，创建一个空的context。意味着将会没有检测能够提供。
public class NullContext extends Context {
public NullContext() {
super(null, "null_context_internal");
}
}
就这样，我了解了context在sentinel中起的作用了。其实，主要就是为了将请求行程有关系的结构，也就是调用链，其结构是由Node（节点）组成的树。
-
entry
Entry 是 Sentinel 中用来表示是否通过限流的一个凭证，就像一个token一样。每次执行 SphU.entry() 或 SphO.entry() 都会返回一个 Entry 给调用者，意思就是告诉调用者，如果正确返回了 Entry 给你，那表示你可以正常访问被 Sentinel 保护的后方服务了，否则 Sentinel 会抛出一个BlockException(如果是 SphO.entry() 会返回false)，这就表示调用者想要访问的服务被保护了，也就是说调用者本身被限流了。
entry中保存了本次执行 entry() 方法的一些基本信息，包括：
- createTime：当前Entry的创建时间，主要用来后期计算rt
- node：当前Entry所关联的node，该node主要是记录了当前context下该资源的统计信息
- origin：当前Entry的调用来源，通常是调用方的应用名称，在 ClusterBuilderSlot.entry() 方法中设置的
- resourceWrapper：当前Entry所关联的资源
当在一个上下文中多次调用了 SphU.entry() 方法时，就会创建一个调用树，这个树的节点之间是通过parent和child关系维持的。
需要注意的是：parent和child是在 CtSph 类的一个私有内部类 CtEntry 中定义的，CtEntry 是 Entry 的一个子类。 由于context中总是保存着调用链树中的当前入口，所以当当前entry执行exit退出时，需要将parent设置为当前入口。
-
node
Node 中保存了资源的实时统计数据，例如：passQps，blockQps，rt等实时数据。正是有了这些统计数据后， Sentinel 才能进行限流、降级等一系列的操作。
node是一个接口，他有一个实现类：StatisticNode，但是StatisticNode本身也有两个子类，一个是DefaultNode，另一个是ClusterNode，DefaultNode又有一个子类叫EntranceNode。
其中entranceNode是每个上下文的入口，该节点是直接挂在root下的，是全局唯一的，每一个context都会对应一个entranceNode。另外defaultNode是记录当前调用的实时数据的，每个defaultNode都关联着一个资源和clusterNode，有着相同资源的defaultNode，他们关联着同一个clusterNode。
首先，我先看看Node的定义。其实想要全部看懂Node的设计，还是应该先了解下Metric度量的设计，否则可能不太明白sentinel怎么进行计算每个时间维度的量：
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
//掌握着资源的时实统计数据
public interface Node extends OccupySupport, DebugSupport {
//一分钟的所有来源请求(pass + block)
long totalRequest();
//一分钟pass的请求
long totalPass();
long totalSuccess();
long blockRequest();
//一分钟异常总数
long totalException();
//每秒通过的请求
double passQps();
double blockQps();
//passQps + blockQps
double totalQps();
//每秒调用了Entry.exit()的请求。调用了这方法说明整个流程走完了，没有抛出blockException也就是没有进行熔断限流
double successQps();
//目前为止预估最大的成功qps
double maxSuccessQps();
double exceptionQps();
//rt - response time。一秒平均返回耗时
double avgRt();
double minRt();
//当前激活中的线程总数
int curThreadNum();
//迁移秒钟被限制的QPS
double previousBlockQps();
double previousPassQps();
//所有资源的可用的检查节点
Map<Long, MetricNode> metrics();
/**
* Fetch all raw metric items that satisfies the time predicate.
*
* @param timePredicate time predicate
* @return raw metric items that satisfies the time predicate
* @since 1.7.0
*/
List<MetricNode> rawMetricsInMin(Predicate<Long> timePredicate);
void addPassRequest(int count);
void addRtAndSuccess(long rt, int success);
void increaseBlockQps(int count);
void increaseExceptionQps(int count);
void increaseThreadNum();
void decreaseThreadNum();
//终止内部计算器。
void reset();
}
Node接口的描述也有了。那么剩下的实现其实就是在其基础上实现不同的功能了。我不打算一一进行分析了。所以，就暂时先分析一个StaticNode吧，毕竟其他的Node也都是继承于他。
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
/*
统计节点有三种实时统计指标：
秒级统计
分钟级别统计
线程总数
sentinel用滑动窗口来记录和统计实时的resource数据。
ArrayMetric的LeapArray是实现滑动窗口的基础设施。
案例1：当第一个请求进来，sentinel会创建一个新的窗口，这个窗口的意思是有着能够将统计数据保存在特定时间跨度的桶。例如总响应时间，QPS，拦截请求等等。这个时间跨度是由样本数量确定的，也就是说是动态的。
* 0 100ms
* +-------+--→ 滑动窗口
* ^
* |
* 请求
Sentinel使用有效桶的静态信息来决定是否可以传递此请求。比如，如果一个规则定义是只能有100个请求能够通过，他会在有效桶中将所有qps加起来与规则定义的阈值（100）进行对比。
案例2：连续的请求。
* 0 100ms 200ms 300ms
* +-------+-------+-------+-----→ 滑动窗口
* ^
* |
* 请求
案例3：请求不间断的到来，并且前一个桶变成了无效。
* 0 100ms 200ms 800ms 900ms 1000ms 1300ms
* +-------+-------+ ...... +-------+-------+ ...... +-------+-----→ 滑动窗口
* ^
* |
* 请求
* 滑动窗口会变成:
* <pre>
* 300ms 800ms 900ms 1000ms 1300ms
* + ...... +-------+ ...... +-------+-----→ 滑动窗口
* ^
* |
* 请求
*/
public class StatisticNode implements Node {
//保存最近N毫秒的统计信息，时间跨度N被给定的样本总数划分
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,
IntervalProperty.INTERVAL);
//保存最近60秒的统计信息，windowLengthInMs被故意设置为1000毫秒，即每秒每个桶，这样我们就可以获得每秒的准确统计数据。
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);
//线程总数
private LongAdder curThreadNum = new LongAdder();
//获取度量的最后一个时间戳。
private long lastFetchTime = -1;
//获取资源的所有有效度量节点。
@Override
public Map<Long, MetricNode> metrics() {
//在单线程调度程序池下，获取操作是线程安全的。
long currentTime = TimeUtil.currentTimeMillis();
currentTime = currentTime - currentTime % 1000;
Map<Long, MetricNode> metrics = new ConcurrentHashMap<>();
List<MetricNode> nodesOfEverySecond = rollingCounterInMinute.details();
long newLastFetchTime = lastFetchTime;
// 迭代所有资源的度量，过滤有效的度量(非空的和最新的)。
for (MetricNode node : nodesOfEverySecond) {
if (isNodeInTime(node, currentTime) && isValidMetricNode(node)) {
metrics.put(node.getTimestamp(), node);
newLastFetchTime = Math.max(newLastFetchTime, node.getTimestamp());
}
}
lastFetchTime = newLastFetchTime;
return metrics;
}
//生成满足时间谓词的聚合度量项。
@Override
public List<MetricNode> rawMetricsInMin(Predicate<Long> timePredicate) {
return rollingCounterInMinute.detailsOnCondition(timePredicate);
}
//度量node时间是否在上次获取时间和当前时间之内
private boolean isNodeInTime(MetricNode node, long currentTime) {
return node.getTimestamp() > lastFetchTime && node.getTimestamp() < currentTime;
}
//判断度量node是否有效
private boolean isValidMetricNode(MetricNode node) {
return node.getPassQps() > 0 || node.getBlockQps() > 0 || node.getSuccessQps() > 0
|| node.getExceptionQps() > 0 || node.getRt() > 0 || node.getOccupiedPassQps() > 0;
}
//重置度量数组
@Override
public void reset() {
rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT, IntervalProperty.INTERVAL);
}
@Override
public long totalRequest() {
return rollingCounterInMinute.pass() + rollingCounterInMinute.block();
}
@Override
public long blockRequest() {
return rollingCounterInMinute.block();
}
@Override
public double blockQps() {
return rollingCounterInSecond.block() / rollingCounterInSecond.getWindowIntervalInSec();
}
@Override
public double previousBlockQps() {
return this.rollingCounterInMinute.previousWindowBlock();
}
@Override
public double previousPassQps() {
return this.rollingCounterInMinute.previousWindowPass();
}
@Override
public double totalQps() {
return passQps() + blockQps();
}
@Override
public long totalSuccess() {
return rollingCounterInMinute.success();
}
@Override
public double exceptionQps() {
return rollingCounterInSecond.exception() / rollingCounterInSecond.getWindowIntervalInSec();
}
@Override
public long totalException() {
return rollingCounterInMinute.exception();
}
@Override
public double passQps() {
return rollingCounterInSecond.pass() / rollingCounterInSecond.getWindowIntervalInSec();
}
@Override
public long totalPass() {
return rollingCounterInMinute.pass();
}
@Override
public double successQps() {
return rollingCounterInSecond.success() / rollingCounterInSecond.getWindowIntervalInSec();
}
@Override
public double maxSuccessQps() {
return (double) rollingCounterInSecond.maxSuccess() * rollingCounterInSecond.getSampleCount()
/ rollingCounterInSecond.getWindowIntervalInSec();
}
@Override
public double occupiedPassQps() {
return rollingCounterInSecond.occupiedPass() / rollingCounterInSecond.getWindowIntervalInSec();
}
@Override
public double avgRt() {
long successCount = rollingCounterInSecond.success();
if (successCount == 0) {
return 0;
}
return rollingCounterInSecond.rt() * 1.0 / successCount;
}
@Override
public double minRt() {
return rollingCounterInSecond.minRt();
}
@Override
public int curThreadNum() {
return (int)curThreadNum.sum();
}
@Override
public void addPassRequest(int count) {
rollingCounterInSecond.addPass(count);
rollingCounterInMinute.addPass(count);
}
@Override
public void addRtAndSuccess(long rt, int successCount) {
rollingCounterInSecond.addSuccess(successCount);
rollingCounterInSecond.addRT(rt);
rollingCounterInMinute.addSuccess(successCount);
rollingCounterInMinute.addRT(rt);
}
@Override
public void increaseBlockQps(int count) {
rollingCounterInSecond.addBlock(count);
rollingCounterInMinute.addBlock(count);
}
@Override
public void increaseExceptionQps(int count) {
rollingCounterInSecond.addException(count);
rollingCounterInMinute.addException(count);
}
@Override
public void increaseThreadNum() {
curThreadNum.increment();
}
@Override
public void decreaseThreadNum() {
curThreadNum.decrement();
}
@Override
public void debug() {
rollingCounterInSecond.debug();
}
//这个是OccupySupport接口的实现。尝试占用有一段时间窗口的令牌。如果占位成功，一个小于occupyTimeout的值会与OccupyTimeoutProperty结构一同返回。具体的解释需要去OccupySupport接口中查看。
@Override
public long tryOccupyNext(long currentTime, int acquireCount, double threshold) {
double maxCount = threshold * IntervalProperty.INTERVAL / 1000;
long currentBorrow = rollingCounterInSecond.waiting();
if (currentBorrow >= maxCount) {
return OccupyTimeoutProperty.getOccupyTimeout();
}
int windowLength = IntervalProperty.INTERVAL / SampleCountProperty.SAMPLE_COUNT;
long earliestTime = currentTime - currentTime % windowLength + windowLength - IntervalProperty.INTERVAL;
int idx = 0;
/*
* Note: here {@code currentPass} may be less than it really is NOW, because time difference
* since call rollingCounterInSecond.pass(). So in high concurrency, the following code may
* lead more tokens be borrowed.
*/
long currentPass = rollingCounterInSecond.pass();
while (earliestTime < currentTime) {
long waitInMs = idx * windowLength + windowLength - currentTime % windowLength;
if (waitInMs >= OccupyTimeoutProperty.getOccupyTimeout()) {
break;
}
long windowPass = rollingCounterInSecond.getWindowPass(earliestTime);
if (currentPass + currentBorrow + acquireCount - windowPass <= maxCount) {
return waitInMs;
}
earliestTime += windowLength;
currentPass -= windowPass;
idx++;
}
return OccupyTimeoutProperty.getOccupyTimeout();
}
@Override
public long waiting() {
return rollingCounterInSecond.waiting();
}
@Override
public void addWaitingRequest(long futureTime, int acquireCount) {
rollingCounterInSecond.addWaiting(futureTime, acquireCount);
}
@Override
public void addOccupiedPass(int acquireCount) {
rollingCounterInMinute.addOccupiedPass(acquireCount);
rollingCounterInMinute.addPass(acquireCount);
}
}
他们之间的继承关系，也侧面反映出来了，每个资源都有一个统计信息，对应存储到Node上，但是上一级的信息会是下一级信息的一个统计。
-
metirc
Metric 是 Sentinel 中用来进行实时数据统计的度量接口，node就是通过metric来进行数据统计的。而metric本身也并没有统计的能力，他也是通过Window来进行统计的。
Metric有一个实现类：ArrayMetric，在ArrayMetric中主要是通过一个叫LeapArray
如果想要了解sentinel是如何计算度量统计数据的，需要深度LeapArray这个类。里面有很多方法值得学习的。我想抽时间学习一下他采用的时间窗口算法，然后再写一篇自己的理解吧。
本文参考：
本文由作者按照
CC BY 4.0
进行授权