在上篇博客中，写了一集群限流的Demo，这篇来分析记录一下集群限流的原理。
不管是集群Client，或者是Server，都会实现TokenService服务，Server端如果是内嵌TokenService服务，则默认使用DefaultEmbeddedTokenServer，而Client端则会使用DefaultClusterTokenClient,类图如下:
diagram.png
import java.util.Collection;
public interface TokenService {
//获取令牌Token， 参数规则Id，获取令牌数，优先级
TokenResult requestToken(Long ruleId, int acquireCount, boolean prioritized);
TokenResult requestParamToken(Long ruleId, int acquireCount, Collection<Object> params);
}
在服务端获取令牌的时候，实质是通过 DefaultEmbeddedTokenServer#requestToken获取Token
public class DefaultEmbeddedTokenServer implements EmbeddedClusterTokenServer {
private final TokenService tokenService = TokenServiceProvider.getService();
private final ClusterTokenServer server = new SentinelDefaultTokenServer(true);
@Override
public TokenResult requestToken(Long ruleId, int acquireCount, boolean prioritized) {
if (tokenService != null) {
return tokenService.requestToken(ruleId, acquireCount, prioritized);
}
return new TokenResult(TokenResultStatus.FAIL);
}
@Override
public TokenResult requestParamToken(Long ruleId, int acquireCount, Collection<Object> params) {
if (tokenService != null) {
return tokenService.requestParamToken(ruleId, acquireCount, params);
}
return new TokenResult(TokenResultStatus.FAIL);
}
}
public class DefaultTokenService implements TokenService {
//获取令牌
@Override
public TokenResult requestToken(Long ruleId, int acquireCount, boolean prioritized) {
//判断是否是有效的请求
if (notValidRequest(ruleId, acquireCount)) {
return badRequest();
}
// 根据RuleId查询FlowRule
FlowRule rule = ClusterFlowRuleManager.getFlowRuleById(ruleId);
if (rule == null) {
return new TokenResult(TokenResultStatus.NO_RULE_EXISTS);
}
//获取令牌
return ClusterFlowChecker.acquireClusterToken(rule, acquireCount, prioritized);
}
//判断是否是一个有效的请求
private boolean notValidRequest(Long id, int count) {
return id == null || id <= 0 || count <= 0;
}
}
//ClusterFlowChecker.java
static TokenResult acquireClusterToken(/*@Valid*/ FlowRule rule, int acquireCount, boolean prioritized) {
Long id = rule.getClusterConfig().getFlowId();
//是否继续，根据RuleId，获取NameSpace，根据nameSpace，判断nameSpace限流是否通过
if (!allowProceed(id)) {
return new TokenResult(TokenResultStatus.TOO_MANY_REQUEST);
}
ClusterMetric metric = ClusterMetricStatistics.getMetric(id);
if (metric == null) {
return new TokenResult(TokenResultStatus.FAIL);
}
//获取Metric，滑动窗口实现，这里获取的是通过的请求平均值
double latestQps = metric.getAvg(ClusterFlowEvent.PASS_REQUEST);
//获取全局阀值 根据规则判断是否为全局限流还是平均分摊，并获取总的阀值
double globalThreshold = calcGlobalThreshold(rule) * ClusterServerConfigManager.getExceedCount();
//判断剩余请求数
double nextRemaining = globalThreshold - latestQps - acquireCount;
//如果>=0，则代表请求可以通过
if (nextRemaining >= 0) {
//记录请求数量
metric.add(ClusterFlowEvent.PASS, acquireCount);
metric.add(ClusterFlowEvent.PASS_REQUEST, 1);
if (prioritized) {
metric.add(ClusterFlowEvent.OCCUPIED_PASS, acquireCount);
}
return new TokenResult(TokenResultStatus.OK)
.setRemaining((int) nextRemaining)
.setWaitInMs(0);
} else {
//这里忽略优先级逻辑
//其他直接返回失败
metric.add(ClusterFlowEvent.BLOCK, acquireCount);
metric.add(ClusterFlowEvent.BLOCK_REQUEST, 1);
ClusterServerStatLogUtil.log("flow|block|" + id, acquireCount);
ClusterServerStatLogUtil.log("flow|block_request|" + id, 1);
if (prioritized) {
metric.add(ClusterFlowEvent.OCCUPIED_BLOCK, acquireCount);
ClusterServerStatLogUtil.log("flow|occupied_block|" + id, 1);
}
return blockedResult();
}
}
static boolean allowProceed(long flowId) {
String namespace = ClusterFlowRuleManager.getNamespace(flowId);
return GlobalRequestLimiter.tryPass(namespace);
}
static boolean allowProceed(long flowId) {
String namespace = ClusterFlowRuleManager.getNamespace(flowId);
return GlobalRequestLimiter.tryPass(namespace);
}
public static boolean tryPass(String namespace) {
if (namespace == null) {
return false;
}
RequestLimiter limiter = GLOBAL_QPS_LIMITER_MAP.get(namespace);
if (limiter == null) {
return true;
}
return limiter.tryPass();
}
public boolean tryPass() {
if (canPass()) {
add(1);
return true;
}
return false;
}
ClusterServerConfigManager.loadGlobalFlowConfig配置了nameSpace对应的ServerFlowConfig
而在客户端的时候，通过netty通信发送到服务端，由服务端验证是否通过。
@Override
public TokenResult requestToken(Long flowId, int acquireCount, boolean prioritized) {
//验证是否有效请求
if (notValidRequest(flowId, acquireCount)) {
return badRequest();
}
//初始化FlowRequest
FlowRequestData data = new FlowRequestData().setCount(acquireCount)
.setFlowId(flowId).setPriority(prioritized);
ClusterRequest<FlowRequestData> request = new ClusterRequest<>(ClusterConstants.MSG_TYPE_FLOW, data);
try {
//发送请求到服务端
TokenResult result = sendTokenRequest(request);
logForResult(result);
return result;
} catch (Exception ex) {
ClusterClientStatLogUtil.log(ex.getMessage());
return new TokenResult(TokenResultStatus.FAIL);
}
}
private TokenResult sendTokenRequest(ClusterRequest request) throws Exception {
if (transportClient == null) {
RecordLog.warn(
"[DefaultClusterTokenClient] Client not created, please check your config for cluster client");
return clientFail();
}
ClusterResponse response = transportClient.sendRequest(request);
TokenResult result = new TokenResult(response.getStatus());
if (response.getData() != null) {
FlowTokenResponseData responseData = (FlowTokenResponseData)response.getData();
result.setRemaining(responseData.getRemainingCount())
.setWaitInMs(responseData.getWaitInMs());
}
return result;
}
在FlowSlot限流的时候，根据节点配置是否启用ClusterMode，判断是否走限流，然后根据节点状态(是Server，或者是Client)获取服务，申请令牌。
static boolean passCheck(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node, int acquireCount,
boolean prioritized) {
String limitApp = rule.getLimitApp();
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 passClusterCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
boolean prioritized) {
try {
//获取令牌服务
TokenService clusterService = pickClusterService();
if (clusterService == null) {
return fallbackToLocalOrPass(rule, context, node, acquireCount, prioritized);
}
long flowId = rule.getClusterConfig().getFlowId();
//申请令牌
TokenResult result = clusterService.requestToken(flowId, acquireCount, prioritized);
return applyTokenResult(result, rule, context, node, acquireCount, prioritized);
} catch (Throwable ex) {
RecordLog.warn("[FlowRuleChecker] Request cluster token unexpected failed", ex);
}
//如果失败，降级为单机模式
return fallbackToLocalOrPass(rule, context, node, acquireCount, prioritized);
}
private static TokenService pickClusterService() {
if (ClusterStateManager.isClient()) {
return TokenClientProvider.getClient();
}
if (ClusterStateManager.isServer()) {
return EmbeddedClusterTokenServerProvider.getServer();
}
return null;
}
网友评论