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;
    }
@Test
    public void testReleaseExpiredToken() throws InterruptedException {
        ConnectionManager.addConnection("test", "127.0.0.1");
        FlowRule rule = ClusterFlowRuleManager.getFlowRuleById(179L);
        for (int i = 0; i < 10; i++) {
            ConcurrentClusterFlowChecker.acquireConcurrentToken("127.0.0.1", rule, 1);
        }
        Thread.sleep(3000);
        Assert.assertTrue("fail to acquire and release token", CurrentConcurrencyManager.get(rule.getClusterConfig().getFlowId()).get() == 0 && TokenCacheNodeManager.getSize() == 0);
    }