Sentinel入门（一） Sentinel在Dubbo的实现

小咪咪 2023-06-28 06:27 35阅读 0赞

### 文章目录 ###

*   *  Demo
     *  Dubbo Consumer Sentinal Filter
     *  Dubbo Provider Sentinal
     *  Sentinel实现原理

## Demo ##

Consumer.java

public static void main(String[] args) { 
    
    		FlowRule flowRule = new FlowRule();
    		flowRule.setResource(
    				"com.alibaba.cloud.examples.FooService:hello(java.lang.String)");
    		// 设置1s内只能调用10次
            flowRule.setCount(10);
    		flowRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
    		flowRule.setLimitApp("default");
            // 注册限流规则
    		FlowRuleManager.loadRules(Collections.singletonList(flowRule));
    
    		SpringApplicationBuilder consumerBuilder = new SpringApplicationBuilder();
    		ApplicationContext applicationContext = consumerBuilder
    				.web(WebApplicationType.NONE).sources(SentinelDubboConsumerApp.class)
    				.run(args);
    
    		FooServiceConsumer service = applicationContext.getBean(FooServiceConsumer.class);
            // 测试调用15次，期望应该是有5调用失败的
    		for (int i = 0; i < 15; i++) { 
    			try { 
    				String message = service.hello("Jim");
    				System.out.println((i + 1) + " -> Success: " + message);
    			}
    			catch (SentinelRpcException ex) { 
    				System.out.println("Blocked");
    			}
    			catch (Exception ex) { 
    				ex.printStackTrace();
    			}
    		}
    	}

FooService服务在Sentinel下对应的资源名是 com.alibaba.cloud.examples.FooService:hello(java.lang.String) 。详细见下面的Filter的`getResourceName`， 定义该资源名对应的限流规则：

public static String getResourceName(Invoker<?> invoker, Invocation invocation) { 
            StringBuilder buf = new StringBuilder(64);
            buf.append(invoker.getInterface().getName())
                .append(":")
                .append(invocation.getMethodName())
                .append("(");
            boolean isFirst = true;
            for (Class<?> clazz : invocation.getParameterTypes()) { 
                if (!isFirst) { 
                    buf.append(",");
                }
                buf.append(clazz.getName());
                isFirst = false;
            }
            buf.append(")");
            return buf.toString();
        }

由于设置的qps是10。调用15次查看是否被限流

1 -> Success: hello, Jim
    2 -> Success: hello, Jim
    3 -> Success: hello, Jim
    4 -> Success: hello, Jim
    5 -> Success: hello, Jim
    6 -> Success: hello, Jim
    7 -> Success: hello, Jim
    8 -> Success: hello, Jim
    9 -> Success: hello, Jim
    10 -> Success: hello, Jim
    Blocked
    Blocked
    Blocked
    Blocked
    Blocked

可以看到是正常被限流的，那么看看是如何限流的。

## Dubbo Consumer Sentinal Filter ##

@Activate(group = "consumer")
    public class SentinelDubboConsumerFilter extends AbstractDubboFilter implements Filter { 
    
        public SentinelDubboConsumerFilter() { 
            RecordLog.info("Sentinel Dubbo consumer filter initialized");
        }
    
        @Override
        public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { 
            Entry interfaceEntry = null;
            Entry methodEntry = null;
            try { 
                String resourceName = getResourceName(invoker, invocation);
                interfaceEntry = SphU.entry(invoker.getInterface().getName(), ResourceTypeConstants.COMMON_RPC,
                    EntryType.OUT);
                methodEntry = SphU.entry(resourceName, ResourceTypeConstants.COMMON_RPC, EntryType.OUT);
    
                Result result = invoker.invoke(invocation);
                if (result.hasException()) { 
                    Throwable e = result.getException();
                    // Record common exception.
                    Tracer.traceEntry(e, interfaceEntry);
                    Tracer.traceEntry(e, methodEntry);
                }
                return result;
            } catch (BlockException e) { 
                return DubboFallbackRegistry.getConsumerFallback().handle(invoker, invocation, e);
            } catch (RpcException e) { 
                Tracer.traceEntry(e, interfaceEntry);
                Tracer.traceEntry(e, methodEntry);
                throw e;
            } finally { 
                if (methodEntry != null) { 
                    methodEntry.exit();
                }
                if (interfaceEntry != null) { 
                    interfaceEntry.exit();
                }
            }
        }
    }

重点是：

1.  String resourceName = getResourceName(invoker, invocation); resource的生成
2.  接口级别的流控： interfaceEntry = SphU.entry(invoker.getInterface().getName(), ResourceTypeConstants.COMMON\_RPC,
3.  方法级别的流控： methodEntry = SphU.entry(resourceName, ResourceTypeConstants.COMMON\_RPC, EntryType.OUT);
4.  EntryType.OUT代表的是出网流量，即对外调用次数，EntryType.IN代表的是入网流量，即被调用次数

## Dubbo Provider Sentinal ##

public class SentinelDubboProviderFilter implements Filter { 
    
        public SentinelDubboProviderFilter() { 
            RecordLog.info("Sentinel Apache Dubbo provider filter initialized");
        }
    
        @Override
        public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { 
            // Get origin caller.
            String application = DubboUtils.getApplication(invocation, "");
    
            Entry interfaceEntry = null;
            Entry methodEntry = null;
            try { 
                String resourceName = DubboUtils.getResourceName(invoker, invocation);
                String interfaceName = invoker.getInterface().getName();
                // Only need to create entrance context at provider side, as context will take effect
                // at entrance of invocation chain only (for inbound traffic).
                ContextUtil.enter(resourceName, application);
                interfaceEntry = SphU.entry(interfaceName, ResourceTypeConstants.COMMON_RPC, EntryType.IN);
                methodEntry = SphU.entry(resourceName, ResourceTypeConstants.COMMON_RPC,
                    EntryType.IN, invocation.getArguments());
    
                Result result = invoker.invoke(invocation);
                if (result.hasException()) { 
                    Throwable e = result.getException();
                    // Record common exception.
                    Tracer.traceEntry(e, interfaceEntry);
                    Tracer.traceEntry(e, methodEntry);
                }
                return result;
            } catch (BlockException e) { 
                return DubboFallbackRegistry.getProviderFallback().handle(invoker, invocation, e);
            } catch (RpcException e) { 
                Tracer.traceEntry(e, interfaceEntry);
                Tracer.traceEntry(e, methodEntry);
                throw e;
            } finally { 
                if (methodEntry != null) { 
                    methodEntry.exit(1, invocation.getArguments());
                }
                if (interfaceEntry != null) { 
                    interfaceEntry.exit();
                }
                ContextUtil.exit();
            }
        }
    }

与消费者主要有以下的不同

1.  ContextUtil.enter(resourceName, application); 因为上下文只会在调用链的入口才生效（对于入站流量）。
2.  对资源的保护方向不一样。

## Sentinel实现原理 ##

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7bKoUjTi-1578412787919)(docs/images/1578410806043.png)\]][img-7bKoUjTi-1578412787919_docs_images_1578410806043.png]

可以联想到我们之前定义的那个规则：

QPS：10

那么可以之间看到`FlowSlot`

public class FlowSlot extends AbstractLinkedProcessorSlot<DefaultNode> { 
    
        private final FlowRuleChecker checker;
    
        public FlowSlot() { 
            this(new FlowRuleChecker());
        }
    
        /** * Package-private for test. * * @param checker flow rule checker * @since 1.6.1 */
        FlowSlot(FlowRuleChecker checker) { 
            AssertUtil.notNull(checker, "flow checker should not be null");
            this.checker = checker;
        }
    
        @Override
        public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                          boolean prioritized, Object... args) throws Throwable { 
            checkFlow(resourceWrapper, context, node, count, prioritized);
    
            fireEntry(context, resourceWrapper, node, count, prioritized, args);
        }
    
        void checkFlow(ResourceWrapper resource, Context context, DefaultNode node, int count, boolean prioritized)
            throws BlockException { 
            // 真正的流控逻辑
            checker.checkFlow(ruleProvider, resource, context, node, count, prioritized);
        }
    
        @Override
        public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) { 
            fireExit(context, resourceWrapper, count, args);
        }
    
        private final Function<String, Collection<FlowRule>> ruleProvider = new Function<String, Collection<FlowRule>>() { 
            @Override
            public Collection<FlowRule> apply(String resource) { 
                // 捞取我们刚才定义的FlowRule
                Map<String, List<FlowRule>> flowRules = FlowRuleManager.getFlowRuleMap();
                return flowRules.get(resource);
            }
        };
    }

--------------------

1.  在多线程情况下，他的访问次数应该是一个`竟态资源`，那么它是怎么处理的呢？

以上两个问题可能需要读代码来解决了。。。。

未完待续。。。。

[img-7bKoUjTi-1578412787919_docs_images_1578410806043.png]: https://img-blog.csdnimg.cn/20200108000055219.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly93dWh1bGFsYS5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70