private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,
    IntervalProperty.INTERVAL);

private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);

private AtomicInteger curThreadNum = new AtomicInteger(0);

public class ArrayMetric implements Metric {

    private final LeapArray<MetricBucket> data;

    public ArrayMetric(int sampleCount, int intervalInMs) {
        this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
    }

    public ArrayMetric(int sampleCount, int intervalInMs, boolean enableOccupy) {
        if (enableOccupy) {
            this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
        } else {
            this.data = new BucketLeapArray(sampleCount, intervalInMs);
        }
    }
    ......
}

public abstract class LeapArray<T> {

    protected int windowLengthInMs;
    protected int sampleCount;
    protected int intervalInMs;

    protected final AtomicReferenceArray<WindowWrap<T>> array;

    // 对于分钟维度的设置，sampleCount 为 60，intervalInMs 为 60 * 1000
    public LeapArray(int sampleCount, int intervalInMs) {
                // 单个窗口长度，这里是 1000ms
        this.windowLengthInMs = intervalInMs / sampleCount;
        // 一轮总时长 60,000 ms
        this.intervalInMs = intervalInMs;
        // 60 个窗口
        this.sampleCount = sampleCount;

        this.array = new AtomicReferenceArray<>(sampleCount);
    }
    // ......
}

public WindowWrap<T> currentWindow(long timeMillis) {
    if (timeMillis < 0) {
        return null;
    }
    // 获取窗口下标
    int idx = calculateTimeIdx(timeMillis);
    // 计算该窗口的理论开始时间
    long windowStart = calculateWindowStart(timeMillis);

    // 嵌套在一个循环中，因为有并发的情况
    while (true) {
        WindowWrap<T> old = array.get(idx);
        if (old == null) {
            // 窗口未实例化的情况，使用一个 CAS 来设置该窗口实例
            WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
            if (array.compareAndSet(idx, null, window)) {
                return window;
            } else {
                // 存在竞争
                Thread.yield();
            }
        } else if (windowStart == old.windowStart()) {
            // 当前数组中的窗口没有过期
            return old;
        } else if (windowStart > old.windowStart()) {
            // 该窗口已过期，重置窗口的值。使用一个锁来控制并发。
            if (updateLock.tryLock()) {
                try {
                    return resetWindowTo(old, windowStart);
                } finally {
                    updateLock.unlock();
                }
            } else {
                Thread.yield();
            }
        } else if (windowStart < old.windowStart()) {
            // 正常情况都不会走到这个分支，异常情况其实就是时钟回拨，这里返回一个 WindowWrap 是容错
            return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
        }
    }
}

public List<T> values(long timeMillis) {
    if (timeMillis < 0) {
        return new ArrayList<T>();
    }
    int size = array.length();
    List<T> result = new ArrayList<T>(size);

    for (int i = 0; i < size; i++) {
        WindowWrap<T> windowWrap = array.get(i);
        // 过滤掉过期数据
        if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {
            continue;
        }
        result.add(windowWrap.value());
    }
    return result;
}

// 判断当前窗口的数据是否是 60 秒内的
public boolean isWindowDeprecated(long time, WindowWrap<T> windowWrap) {
    return time - windowWrap.windowStart() > intervalInMs;
}

try (Entry entry = SphU.entry("HelloWorld")) {
    // Your business logic here.
    System.out.println("hello world");
} catch (BlockException e) {
    // Handle rejected request.
    e.printStackTrace();
}

ContextUtil.enter("user-center", "app-A");

private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
    throws BlockException {
    // 从 ThreadLocal 中获取 Context 实例
    Context context = ContextUtil.getContext();
    // 如果是 NullContext，那么说明 context name 超过了 2000 个，参见 ContextUtil#trueEnter
    // 这个时候，Sentinel 不再接受处理新的 context 配置，也就是不做这些新的接口的统计、限流熔断等
    if (context instanceof NullContext) {
        return new CtEntry(resourceWrapper, null, context);
    }

    // 我们前面说了，如果我们不显式调用 ContextUtil#enter，这里会进入到默认的 context 中
    if (context == null) {
        context = MyContextUtil.myEnter(Constants.CONTEXT_DEFAULT_NAME, "", resourceWrapper.getType());
    }

    // Sentinel 的全局开关，Sentinel 提供了接口让用户可以在 dashboard 开启/关闭
    if (!Constants.ON) {
        return new CtEntry(resourceWrapper, null, context);
    }

    // 设计模式中的责任链模式。
    // 下面这行代码用于构建一个责任链，入参是 resource，前面我们说过资源的唯一标识是 resource name
    ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);

    // 根据 lookProcessChain 方法，我们知道，当 resource 超过 Constants.MAX_SLOT_CHAIN_SIZE，
    // 也就是 6000 的时候，Sentinel 开始不处理新的请求，这么做主要是为了 Sentinel 的性能考虑
    if (chain == null) {
        return new CtEntry(resourceWrapper, null, context);
    }

    // 执行这个责任链。如果抛出 BlockException，说明链上的某一环拒绝了该请求，
    // 把这个异常往上层业务层抛，业务层处理 BlockException 应该进入到熔断降级逻辑中
    Entry e = new CtEntry(resourceWrapper, chain, context);
    try {
        chain.entry(context, resourceWrapper, null, count, prioritized, args);
    } catch (BlockException e1) {
        e.exit(count, args);
        throw e1;
    } catch (Throwable e1) {
        // This should not happen, unless there are errors existing in Sentinel internal.
        RecordLog.info("Sentinel unexpected exception", e1);
    }
    return e;
}

public class DefaultSlotChainBuilder implements SlotChainBuilder {

    @Override
    public ProcessorSlotChain build() {
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();
        chain.addLast(new NodeSelectorSlot());
        chain.addLast(new ClusterBuilderSlot());
        chain.addLast(new LogSlot());
        chain.addLast(new StatisticSlot());
        chain.addLast(new AuthoritySlot());
        chain.addLast(new SystemSlot());
        chain.addLast(new FlowSlot());
        chain.addLast(new DegradeSlot());
        return chain;
    }
}

// key 是 context name, value 是 DefaultNode 实例
private volatile Map<String, DefaultNode> map = new HashMap<String, DefaultNode>(10);

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
    throws Throwable {
    DefaultNode node = map.get(context.getName());
    if (node == null) {
        synchronized (this) {
            node = map.get(context.getName());
            if (node == null) {
                node = new DefaultNode(resourceWrapper, null);
                HashMap<String, DefaultNode> cacheMap = new HashMap<String, DefaultNode>(map.size());
                cacheMap.putAll(map);
                cacheMap.put(context.getName(), node);
                map = cacheMap;
                // Build invocation tree
                ((DefaultNode) context.getLastNode()).addChild(node);
            }

        }
    }

    context.setCurNode(node);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

if (!"".equals(context.getOrigin())) {
    Node originNode = node.getClusterNode().getOrCreateOriginNode(context.getOrigin());
    context.getCurEntry().setOriginNode(originNode);
}

OperatingSystemMXBean osBean = ManagementFactory.getPlatformMXBean(OperatingSystemMXBean.class);

currentLoad = osBean.getSystemLoadAverage();

currentCpuUsage = osBean.getSystemCpuLoad();

if (cut.compareAndSet(false, true)) {
    ResetTask resetTask = new ResetTask(this);
    pool.schedule(resetTask, timeWindow, TimeUnit.SECONDS);
}

<dependency>
   <groupId>com.alibaba.csp</groupId>
   <artifactId>sentinel-transport-simple-http</artifactId>
   <version>1.6.3</version>
</dependency>

-Dcsp.sentinel.dashboard.server=127.0.0.1:8080 -Dproject.name=sentinel-learning

public static Entry entry(String name) throws BlockException {
    return Env.sph.entry(name, EntryType.OUT, 1, OBJECTS0);
}

public class Env {
    public static final Sph sph = new CtSph();
    static {
        // If init fails, the process will exit.
        InitExecutor.doInit();
    }
}

public static void doInit() {
    if (!initialized.compareAndSet(false, true)) {
        return;
    }
    try {
        ServiceLoader<InitFunc> loader = ServiceLoader.load(InitFunc.class);
        List<OrderWrapper> initList = new ArrayList<OrderWrapper>();
        for (InitFunc initFunc : loader) {
            insertSorted(initList, initFunc);
        }
        for (OrderWrapper w : initList) {
            w.func.init();
        }
        // ...
}

@Override
public void init() {
    HeartbeatSender sender = HeartbeatSenderProvider.getHeartbeatSender();
    if (sender == null) {
        RecordLog.warn("[HeartbeatSenderInitFunc] WARN: No HeartbeatSender loaded");
        return;
    }

    initSchedulerIfNeeded();
    long interval = retrieveInterval(sender);
    setIntervalIfNotExists(interval);
    // 启动一个定时器，发送心跳信息
    scheduleHeartbeatTask(sender, interval);
}

// 2 个时间窗口，每个窗口长度 0.5 秒
public ArrayMetric(int sampleCount, int intervalInMs) {
    this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
}

@Override
public boolean isWindowDeprecated(long time, WindowWrap<MetricBucket> windowWrap) {
    // Tricky: will only calculate for future.
    return time >= windowWrap.windowStart();
}

@Override
protected WindowWrap<MetricBucket> resetWindowTo(WindowWrap<MetricBucket> w, long time) {
    // Update the start time and reset value.
    w.resetTo(time);
    MetricBucket borrowBucket = borrowArray.getWindowValue(time);
    if (borrowBucket != null) {
        w.value().reset();
        w.value().addPass((int)borrowBucket.pass());
    } else {
        w.value().reset();
    }
    return w;
}

@Override
public void addWaiting(long time, int acquireCount) {
    WindowWrap<MetricBucket> window = borrowArray.currentWindow(time);
    window.value().add(MetricEvent.PASS, acquireCount);
}

@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
    int curCount = avgUsedTokens(node);
    if (curCount + acquireCount > count) {
        // 只有设置了 prioritized 的情况才会进入到下面的 if 分支
        // 也就是说，对于一般的场景，被限流了，就快速失败
        if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {
            long currentTime;
            long waitInMs;
            currentTime = TimeUtil.currentTimeMillis();
            // 下面的这行 tryOccupyNext 非常复杂，大意就是说去占有"未来的"令牌
            // 可以看到，下面做了 sleep，为了保证 QPS 不会因为预占而撑大
            waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);
            if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {
                // 就是这里设置了 borrowArray 的值
                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;
}

public static void main(String[] args) {
    // 下面几行代码设置了 QPS 阈值是 100
    FlowRule rule = new FlowRule("test");
    rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
    rule.setCount(100);
    rule.setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_DEFAULT);
    List<FlowRule> list = new ArrayList<>();
    list.add(rule);
    FlowRuleManager.loadRules(list);

    // 先通过一个请求，让 clusterNode 先建立起来
    try (Entry entry = SphU.entry("test")) {
    } catch (BlockException e) {
    }

    // 起一个线程一直打印 qps 数据
    new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                System.out.println(ClusterBuilderSlot.getClusterNode("test").passQps());
            }
        }
    }).start();

    while (true) {
        try (Entry entry = SphU.entry("test")) {
            Thread.sleep(5);
        } catch (BlockException e) {
            // ignore
        } catch (InterruptedException e) {
            // ignore
        }
    }
}