[feature]finish cluster concurrent flow control rule checker

[yunfeiyanggzq]

Signed-off-by: yunfeiyanggzq [email protected]

Describe what this PR does / why we need it

完成集群并发流控流控规则源

Does this pull request fix one issue?

Fixes #1629 此处有项目整体代码和测试说明

Describe how you did it

本pr是集群并发流控的一部分代码，主要完成DefaultTokenService,本代码并没有接入FlowSlot的FlowCheker中，所以对现有的业务并没有发生任何的改变。主要的功能是实现了集群并发流控的规则源,不能进行netty通信

改造sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/ClusterFlowConfig.java，调整集群流控规则，增加并发流控的相应控制参数
改造sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/flow/rule/ClusterFlowRuleManager.java 主要适配并发集群流控中的一些参数检测。
新增sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/flow/statistic/concurrent/CurrentConcurrencyManager.java 维护每个rule所对应的当前并发量
新增sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/flow/statistic/concurrent/TokenCacheNodeManager.java 存储目前的token信息
新增sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/flow/statistic/concurrent/expire/RegularExpireStrategy.java 删除过期token
新增sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/flow/ConcurrentClusterFlowChecker.java 主要对token是否能分发进行检测和回收token

server端新增接口

 /**
     * Request acquire concurrent tokens from remote token server.
     *
     * @param clientAddress the address of the request belong.
     * @param ruleId ruleId the unique rule ID
     * @param acquireCount token count to acquire
     * @return result of the token request
     */
    TokenResult requestConcurrentToken(String clientAddress,Long ruleId,int acquireCount);
    /**
     * Request lease concurrent tokens from remote token server.
     *
     * @param tokenId the unique token ID
     */
    void releaseConcurrentToken(Long tokenId);

Describe how to verify it

查看单元测试。

Special notes for reviews

[sczyh30]

We may need to pay attention to the memory and GC footprint when there are large amount of requests coming.

[yunfeiyanggzq]

the max memory used is decided by rules*maxConcurrency, I think memory used is controllable.every token information is small and the token can be replaced fastly during the process of acquiring and releasing, there may be some young gc in the case you mention, I think it can be acceptable

[sczyh30]

We may provide benchmark of the scenario. Frequent YGC may need optimizing.

[yunfeiyanggzq]

I will do it, Thanks !

[yunfeiyanggzq]

Because the process is too short, so the image is not statistically accurate.

the code is as follows:

       @Before
    public void setUp() {
        FlowRule rule = new FlowRule();
        ClusterFlowConfig config = new ClusterFlowConfig();
        config.setResourceTimeout(500);
        config.setClientOfflineTime(1000);
        config.setFlowId(111L);
        config.setThresholdType(ClusterRuleConstant.FLOW_THRESHOLD_GLOBAL);
        rule.setClusterConfig(config);
        rule.setClusterMode(true);
        rule.setCount(1000);
        rule.setResource("test");
        rule.setGrade(RuleConstant.FLOW_GRADE_THREAD);
        ArrayList<FlowRule> rules = new ArrayList<>();
        rules.add(rule);
        ClusterFlowRuleManager.registerPropertyIfAbsent("1-name");
        ClusterFlowRuleManager.loadRules("1-name", rules);
    }
   
   @Test
    public void testConcurrentAcquireAndRelease() throws InterruptedException {
        setCurrentMillis(System.currentTimeMillis());
        final FlowRule rule = ClusterFlowRuleManager.getFlowRuleById(111L);
        final CountDownLatch countDownLatch = new CountDownLatch(1000000);
        ExecutorService pool = Executors.newFixedThreadPool(100);
        final AtomicInteger success=new AtomicInteger(0);
        for (long i = 0; i < 1000000; i++) {
            Runnable task = new Runnable() {
                @Override
                public void run() {
                    assert rule != null;
                    TokenResult result = ConcurrentClusterFlowChecker.acquireConcurrentToken("127.0.0.1", rule, 1);
                    Assert.assertTrue("concurrent control fail", CurrentConcurrencyManager.get(111L).get() <= rule.getCount());
                    if (result.getStatus() == TokenResultStatus.OK) {
                        success.incrementAndGet();
                        ConcurrentClusterFlowChecker.releaseConcurrentToken(result.getTokenId());
                    }
                    countDownLatch.countDown();
                }
            };
            pool.execute(task);
        }
        countDownLatch.await();
        pool.shutdown();
        System.out.println(success.get()+"成功的");
        assert rule != null;
        Assert.assertTrue("fail to acquire and release token",
                CurrentConcurrencyManager.get(rule.getClusterConfig().getFlowId()).get() == 0 && TokenCacheNodeManager.getSize() == 0);
    }

[yunfeiyanggzq]

每秒请求量越小gc越少

[sczyh30]

And some of the logs should be improved (e.g. the scope, content).

[yunfeiyanggzq]

And some of the logs should be improved (e.g. the scope, content).

I will do with the review as soon as possible

[yunfeiyanggzq]

I have modify the code with the opinion of @sczyh30 ,please review again @sczyh30 @jasonjoo2010 @cdfive ,Thanks!

[sczyh30]

LGTM. We may improve the commits later.

[sczyh30]

Nice work. Thanks for contributing!