标签：node 降级 return prioritized 流控 int context Sentinel
流控
请使用 IDEA 工具 结合查看源码
FlowSlot 负责流量控制的 ，他的下一个 Slot 是：DegradeSlot （降级）
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
// 检测并应用流控规则
checkFlow(resourceWrapper, context, node, count, prioritized);
// 触发下一个Slot
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);
}
.
private final Function<String, Collection<FlowRule>> ruleProvider = new Function<String, Collection<FlowRule>>() {
@Override
public Collection<FlowRule> apply(String resource) {
// Flow rule map should not be null.
// 获取到所有资源的流控规则
// map的key为资源名称，value为该资源上加载的所有流控规则
// 配置使用是调用方法：FlowRuleManager.loadRules(xxx);目的就是将自定义的流控规则载入，
// 此处FlowRuleManager.getFlowRuleMap() 将所有流控规则取出
Map<String, List<FlowRule>> flowRules = FlowRuleManager.getFlowRuleMap();
// 获取指定资源的所有流控规则
return flowRules.get(resource);
}
};
.
public void checkFlow(Function<String, Collection<FlowRule>> ruleProvider, ResourceWrapper resource,
Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {
if (ruleProvider == null || resource == null) {
return;
}
// 获取到指定资源的所有流控规则
Collection<FlowRule> rules = ruleProvider.apply(resource.getName());
if (rules != null) {
// 逐个应用流控规则。若无法通过则抛出异常，后续规则不再应用
for (FlowRule rule : rules) {
if (!canPassCheck(rule, context, node, count, prioritized)) {
throw new FlowException(rule.getLimitApp(), rule);
}
}
}
}
.
public boolean canPassCheck(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node,
int acquireCount) {
return canPassCheck(rule, context, node, acquireCount, false);
}
public boolean canPassCheck(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node, int acquireCount,
boolean prioritized) {
// 从规则中获取要限定的来源
String limitApp = rule.getLimitApp();
// 若限流的来源为null，则请求直接通过
if (limitApp == null) {
return true;
}
// 使用规则处理集群流控
if (rule.isClusterMode()) {
return passClusterCheck(rule, context, node, acquireCount, prioritized);
}
// 使用规则处理单机流控
return passLocalCheck(rule, context, node, acquireCount, prioritized);
}
private static boolean passLocalCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
boolean prioritized) {
// 通过规则形成选择出的规则node
Node selectedNode = selectNodeByRequesterAndStrategy(rule, context, node);
// 若没有选择出node，说明没有规则，则直接返回true，表示通过检测
if (selectedNode == null) {
return true;
}
// 使用规则进行逐项检测
return rule.getRater().canPass(selectedNode, acquireCount, prioritized);
}
四个实现类，正是管理控制台中的流控效果
DefaultController 为快速失败实现方案
// 快速失败的流控效果中的通过性判断
@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
// 获取当前时间窗中已经统计的数据
int curCount = avgUsedTokens(node);
// 若已经统计的数据与本次请求的数量和 大于 设置的阈值，则返回false，表示没有通过检测
// 若小于等于阈值，则返回true，表示通过检测
if (curCount + acquireCount > count) {
if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {
long currentTime;
long waitInMs;
currentTime = TimeUtil.currentTimeMillis();
waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);
if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {
node.addWaitingRequest(currentTime + waitInMs, acquireCount);
node.addOccupiedPass(acquireCount);
sleep(waitInMs);
// PriorityWaitException indicates that the request will pass after waiting for {@link @waitInMs}.
throw new PriorityWaitException(waitInMs);
}
}
return false;
}
return true;
}
private int avgUsedTokens(Node node) {
// 若没有选择出node，则说明没有做统计工作，直接返回0
if (node == null) {
return DEFAULT_AVG_USED_TOKENS;
}
// 若阈值类型为线程数，则直接返回当前的线程数量；
// 若阈值类型为QPS，则返回统计的当前的QPS
return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
}
StatisticNode 中的实现
@Override
public double passQps() {
// rollingCounterInSecond.pass() 当前时间窗中统计的通过的请求数量
// rollingCounterInSecond.getWindowIntervalInSec() 时间窗长度
// 这两个数相除，计算出的就是QPS
return rollingCounterInSecond.pass() / rollingCounterInSecond.getWindowIntervalInSec();
}
流控规则对象，管理控制台页面设置流控格则中封装的就是这个对象
public class FlowRule extends AbstractRule {
public FlowRule() {
super();
setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);
}
public FlowRule(String resourceName) {
super();
setResource(resourceName);
setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);
}
/**
* The threshold type of flow control (0: thread count, 1: QPS).
*/
// 阈值类型，0表示线程数限流，1表示QPS限流
private int grade = RuleConstant.FLOW_GRADE_QPS;
/**
* Flow control threshold count.
*/
// 阈值
private double count;
/**
* 流控模式
* Flow control strategy based on invocation chain.
* 直接流控
* {@link RuleConstant#STRATEGY_DIRECT} for direct flow control (by origin);
* 关联流控
* {@link RuleConstant#STRATEGY_RELATE} for relevant flow control (with relevant resource);
* 链路流控
* {@link RuleConstant#STRATEGY_CHAIN} for chain flow control (by entrance resource).
*/
private int strategy = RuleConstant.STRATEGY_DIRECT;
/**
* Reference resource in flow control with relevant resource or context.
*/
// 流控模式为关联流控时的关联资源
private String refResource;
/**
* Rate limiter control behavior.
* 0. default(reject directly), 1. warm up, 2. rate limiter, 3. warm up + rate limiter
*/
// 流控效果：0表示快速失败，1表示warm up（令牌桶算法），2表示排队等待（漏斗算法），3表示warm up+排队等待
private int controlBehavior = RuleConstant.CONTROL_BEHAVIOR_DEFAULT;
// warm up预热时长
private int warmUpPeriodSec = 10;
/**
* Max queueing time in rate limiter behavior.
*/
// 排队等待的超时时间
private int maxQueueingTimeMs = 500;
// 是否集群模式
private boolean clusterMode;
。。。。。。。。。。。。。。
降级 ( DegradeSlot )
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
// 完成熔断降级检测
performChecking(context, resourceWrapper);
// 触发下一个节点
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
void performChecking(Context context, ResourceWrapper r) throws BlockException {
// 获取到当前资源的所有熔断器
List<CircuitBreaker> circuitBreakers = DegradeRuleManager.getCircuitBreakers(r.getName());
// 若熔断器为空，则直结束
if (circuitBreakers == null || circuitBreakers.isEmpty()) {
return;
}
// 逐个尝试所有熔断器
for (CircuitBreaker cb : circuitBreakers) {
// 若没有通过当前熔断器，则直接抛出异常
if (!cb.tryPass(context)) {
throw new DegradeException(cb.getRule().getLimitApp(), cb.getRule());
}
}
}
.
@Override
public boolean tryPass(Context context) {
// Template implementation.
// 熔断器状态为关闭状态，则请求可以通过
if (currentState.get() == State.CLOSED) {
return true;
}
// 熔断器状态为打开状态，此时再查看，
// 若下次时间窗时间点已经到达，且熔断器成功由Open变为了Half-Open，则请求通过
if (currentState.get() == State.OPEN) {
// For half-open state we allow a request for probing.
return retryTimeoutArrived() && fromOpenToHalfOpen(context);
}
return false;
}
>>>>>>> 接下一篇：https://www.cnblogs.com/Alay/p/15488154.html <<<<<<<<<<<<<
标签：node,降级,return,prioritized,流控,int,context,Sentinel 来源： https://www.cnblogs.com/Alay/p/15488148.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。