-
Notifications
You must be signed in to change notification settings - Fork 8.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: yunfeiyanggzq <yunfeiyang@buaa.edu.cn>
- Loading branch information
1 parent
17c3ff7
commit 6d8cc45
Showing
24 changed files
with
1,276 additions
and
50 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
102 changes: 102 additions & 0 deletions
102
...ult/src/main/java/com/alibaba/csp/sentinel/cluster/flow/ConcurrentClusterFlowChecker.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| /* | ||
| * Copyright 1999-2018 Alibaba Group Holding Ltd. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.alibaba.csp.sentinel.cluster.flow; | ||
|
|
||
| import com.alibaba.csp.sentinel.cluster.TokenResult; | ||
| import com.alibaba.csp.sentinel.cluster.TokenResultStatus; | ||
| import com.alibaba.csp.sentinel.cluster.flow.rule.ClusterFlowRuleManager; | ||
| import com.alibaba.csp.sentinel.cluster.flow.statistic.concurrent.CurrentConcurrencyManager; | ||
| import com.alibaba.csp.sentinel.cluster.flow.statistic.concurrent.TokenCacheNode; | ||
| import com.alibaba.csp.sentinel.cluster.flow.statistic.concurrent.TokenCacheNodeManager; | ||
| import com.alibaba.csp.sentinel.cluster.server.log.ClusterServerStatLogUtil; | ||
| import com.alibaba.csp.sentinel.log.RecordLog; | ||
| import com.alibaba.csp.sentinel.slots.block.ClusterRuleConstant; | ||
| import com.alibaba.csp.sentinel.slots.block.flow.FlowRule; | ||
|
|
||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| /** | ||
| * @author yunfeiyanggzq | ||
| */ | ||
| final public class ConcurrentClusterFlowChecker { | ||
|
|
||
| public static double calcGlobalThreshold(FlowRule rule) { | ||
| double count = rule.getCount(); | ||
| switch (rule.getClusterConfig().getThresholdType()) { | ||
| case ClusterRuleConstant.FLOW_THRESHOLD_GLOBAL: | ||
| return count; | ||
| case ClusterRuleConstant.FLOW_THRESHOLD_AVG_LOCAL: | ||
| default: | ||
| int connectedCount = ClusterFlowRuleManager.getConnectedCount(rule.getClusterConfig().getFlowId()); | ||
| return count * connectedCount; | ||
| } | ||
| } | ||
|
|
||
| public static TokenResult acquireConcurrentToken(/*@Valid*/ String clientAddress,FlowRule rule, int acquireCount) { | ||
| long flowId = rule.getClusterConfig().getFlowId(); | ||
| AtomicInteger nowCalls = CurrentConcurrencyManager.get(flowId); | ||
| if (nowCalls == null) { | ||
| RecordLog.warn("[ConcurrentClusterFlowChecker] Fail to get nowCalls by flowId<{}>", flowId); | ||
| return new TokenResult(TokenResultStatus.FAIL); | ||
| } | ||
|
|
||
| // check before enter the lock to improve the efficiency | ||
| if (nowCalls.get() + acquireCount > calcGlobalThreshold(rule)) { | ||
| ClusterServerStatLogUtil.log("concurrent|block|" + flowId, acquireCount); | ||
| return new TokenResult(TokenResultStatus.BLOCKED); | ||
| } | ||
|
|
||
| // ensure the atomicity of operations | ||
| // lock different nowCalls to improve the efficiency | ||
| synchronized (nowCalls) { | ||
| // check again whether the request can pass. | ||
| if (nowCalls.get() + acquireCount > calcGlobalThreshold(rule)) { | ||
| ClusterServerStatLogUtil.log("concurrent|block|" + flowId, acquireCount); | ||
| return new TokenResult(TokenResultStatus.BLOCKED); | ||
| } else { | ||
| nowCalls.getAndAdd(acquireCount); | ||
| } | ||
| } | ||
| ClusterServerStatLogUtil.log("concurrent|pass|" + flowId, acquireCount); | ||
| TokenCacheNode node = TokenCacheNode.generateTokenCacheNode(rule, acquireCount, clientAddress); | ||
| TokenCacheNodeManager.putTokenCacheNode(node.getTokenId(), node); | ||
| TokenResult tokenResult = new TokenResult(TokenResultStatus.OK); | ||
| tokenResult.setTokenId(node.getTokenId()); | ||
| return tokenResult; | ||
| } | ||
|
|
||
| public static TokenResult releaseConcurrentToken(/*@Valid*/ long tokenId) { | ||
| TokenCacheNode node = TokenCacheNodeManager.getTokenCacheNode(tokenId); | ||
| if (node == null) { | ||
| RecordLog.info("[ConcurrentClusterFlowChecker] Token<{}> is already released", tokenId); | ||
| return new TokenResult(TokenResultStatus.ALREADY_RELEASE); | ||
| } | ||
| FlowRule rule = ClusterFlowRuleManager.getFlowRuleById(node.getFlowId()); | ||
| if (rule == null) { | ||
| RecordLog.info("[ConcurrentClusterFlowChecker] Fail to get rule by flowId<{}>", node.getFlowId()); | ||
| return new TokenResult(TokenResultStatus.NO_RULE_EXISTS); | ||
| } | ||
| if (TokenCacheNodeManager.removeTokenCacheNode(tokenId) == null) { | ||
| RecordLog.info("[ConcurrentClusterFlowChecker] Token<{}> is already released", tokenId); | ||
| return new TokenResult(TokenResultStatus.ALREADY_RELEASE); | ||
| } | ||
| int acquireCount = node.getAcquireCount(); | ||
| AtomicInteger nowCalls = CurrentConcurrencyManager.get(node.getFlowId()); | ||
| nowCalls.getAndAdd(-1 * acquireCount); | ||
| ClusterServerStatLogUtil.log("concurrent|release|" + rule.getClusterConfig().getFlowId(), acquireCount); | ||
| return new TokenResult(TokenResultStatus.RELEASE_OK); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
52 changes: 52 additions & 0 deletions
52
...a/csp/sentinel/cluster/flow/statistic/concurrent/ClusterConcurrentCheckerLogListener.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| /* | ||
| * Copyright 1999-2018 Alibaba Group Holding Ltd. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.alibaba.csp.sentinel.cluster.flow.statistic.concurrent; | ||
|
|
||
| import com.alibaba.csp.sentinel.cluster.flow.ConcurrentClusterFlowChecker; | ||
| import com.alibaba.csp.sentinel.cluster.flow.rule.ClusterFlowRuleManager; | ||
| import com.alibaba.csp.sentinel.cluster.server.log.ClusterServerStatLogUtil; | ||
| import com.alibaba.csp.sentinel.log.RecordLog; | ||
| import com.alibaba.csp.sentinel.slots.block.flow.FlowRule; | ||
|
|
||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| /** | ||
| * @author yunfeiyanggzq | ||
| */ | ||
| public class ClusterConcurrentCheckerLogListener implements Runnable { | ||
| @Override | ||
| public void run() { | ||
| try { | ||
| collectInformation(); | ||
| } catch (Exception e) { | ||
| RecordLog.warn("[ClusterConcurrentCheckerLogListener] Failed to record concurrent flow control regularly", e); | ||
| } | ||
| } | ||
|
|
||
| private void collectInformation() { | ||
| ConcurrentHashMap<Long, AtomicInteger> nowCallsMap = CurrentConcurrencyManager.getConcurrencyMap(); | ||
| for (long flowId : nowCallsMap.keySet()) { | ||
| FlowRule rule = ClusterFlowRuleManager.getFlowRuleById(flowId); | ||
| if (rule == null || nowCallsMap.get(flowId).get() == 0) { | ||
| continue; | ||
| } | ||
| double concurrencyLevel = ConcurrentClusterFlowChecker.calcGlobalThreshold(rule); | ||
| String resource = rule.getResource(); | ||
| ClusterServerStatLogUtil.log("concurrent|resource:" + resource + "|flowId:" + flowId + "|concurrencyLevel:" + concurrencyLevel, nowCallsMap.get(flowId).get()); | ||
| } | ||
| } | ||
| } |
97 changes: 97 additions & 0 deletions
97
...com/alibaba/csp/sentinel/cluster/flow/statistic/concurrent/CurrentConcurrencyManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| /* | ||
| * Copyright 1999-2018 Alibaba Group Holding Ltd. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.alibaba.csp.sentinel.cluster.flow.statistic.concurrent; | ||
|
|
||
| import com.alibaba.csp.sentinel.concurrent.NamedThreadFactory; | ||
|
|
||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.ScheduledExecutorService; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| /** | ||
| * We use a ConcurrentHashMap<long, AtomicInteger> type structure to store nowCalls corresponding to | ||
| * rules, where the key is flowId and the value is nowCalls. Because nowCalls may be accessed and | ||
| * modified by multiple threads, we consider to design it as an AtomicInteger class . Each newly | ||
| * created rule will add a nowCalls object to this map. If the concurrency corresponding to a rule changes, | ||
| * we will update the corresponding nowCalls in real time. Each request to obtain a token will increase the nowCalls; | ||
| * and the request to release the token will reduce the nowCalls. | ||
| * | ||
| * @author yunfeiyanggzq | ||
| */ | ||
| public final class CurrentConcurrencyManager { | ||
| /** | ||
| * use ConcurrentHashMap to store the nowCalls of rules. | ||
| */ | ||
| private static final ConcurrentHashMap<Long, AtomicInteger> NOW_CALLS_MAP = new ConcurrentHashMap<Long, AtomicInteger>(); | ||
|
|
||
| private static final ScheduledExecutorService SCHEDULER = Executors.newScheduledThreadPool(1, | ||
| new NamedThreadFactory("sentinel-cluster-concurrency-record-task", true)); | ||
|
|
||
| static { | ||
| ClusterConcurrentCheckerLogListener logTask = new ClusterConcurrentCheckerLogListener(); | ||
| SCHEDULER.scheduleAtFixedRate(logTask, 0, 1, TimeUnit.SECONDS); | ||
| } | ||
|
|
||
| /** | ||
| * update current concurrency. | ||
| */ | ||
| public static Boolean update(Long flowId, Integer count) { | ||
|
|
||
| AtomicInteger nowCalls = NOW_CALLS_MAP.get(flowId); | ||
| if (nowCalls == null) { | ||
| return false; | ||
| } | ||
| nowCalls.getAndAdd(count); | ||
| return true; | ||
| } | ||
|
|
||
| /** | ||
| * get the current concurrency. | ||
| */ | ||
| public static AtomicInteger get(Long flowId) { | ||
| return NOW_CALLS_MAP.get(flowId); | ||
| } | ||
|
|
||
| /** | ||
| * delete the current concurrency. | ||
| */ | ||
| public static void remove(Long flowId) { | ||
| NOW_CALLS_MAP.remove(flowId); | ||
| } | ||
|
|
||
| /** | ||
| * add the current concurrency. | ||
| */ | ||
| public static void put(Long flowId, Integer nowCalls) { | ||
| NOW_CALLS_MAP.put(flowId, new AtomicInteger(nowCalls)); | ||
| } | ||
|
|
||
| /** | ||
| * check flow id. | ||
| */ | ||
| public static boolean containsFlowId(Long flowId) { | ||
| return NOW_CALLS_MAP.containsKey(flowId); | ||
| } | ||
|
|
||
| /** | ||
| * get NOW_CALLS_MAP. | ||
| */ | ||
| public static ConcurrentHashMap<Long, AtomicInteger> getConcurrencyMap() { | ||
| return NOW_CALLS_MAP; | ||
| } | ||
| } |
Oops, something went wrong.