归档
- GitHub: Ali-Sentinel-节点与度量
作用
- 保存资源的实时统计信息
节点
节点-类结构
com.alibaba.csp.sentinel.slots.statistic.metric.DebugSupport
java">/** 调试支持 */
public interface DebugSupport {void debug(); // 打印统计信息
}
com.alibaba.csp.sentinel.node.OccupySupport
java">/** 占用支持 */
public interface OccupySupport {...
}
com.alibaba.csp.sentinel.node.Node
java">/** 节点 (用于统计) */
public interface Node extends OccupySupport, DebugSupport {long totalRequest(); // 获取每分钟传入的请求long totalPass(); // 获取每分钟的通过次数long totalSuccess(); // 每分钟完成的请求(Entry.exit())总数long blockRequest(); // 获取每分钟阻止的请求计数long totalException(); // 获取每分钟的异常计数double passQps(); // 获取每秒已通过请求的 QPSdouble blockQps(); // 获取每秒被阻止请求的 QPSdouble totalQps(); // 获取每秒总请求的 QPSdouble successQps(); // 获取每秒已完成请求的(Entry.exit()) QPSdouble maxSuccessQps(); // 获得最大已完成请求的 QPSdouble exceptionQps(); // 发生异常的 QPSdouble avgRt(); // 平均每秒响应时间 (Rt: response time)double minRt();int curThreadNum();double previousBlockQps();double previousPassQps();void addPassRequest(int count);void addRtAndSuccess(long rt, int success);void increaseBlockQps(int count);void increaseExceptionQps(int count);void increaseThreadNum();void decreaseThreadNum();void reset();
}
com.alibaba.csp.sentinel.node.StatisticNode
java">/** 统计节点 */
public class StatisticNode implements Node {// 保存最近 INTERVAL(1000) 毫秒的统计信息。// 节点的统计逻辑都委派给其处理，ref: sign_i_100 & sign_c_100private transient volatile Metric rollingCounterInSecond = new ArrayMetric( // sign_cm_101SampleCountProperty.SAMPLE_COUNT, // def: 2IntervalProperty.INTERVAL // def: 1000);// 保存最近 60 秒的统计信息。// 节点的统计逻辑都委派给其处理，ref: sign_i_100 & sign_c_100private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false); // sign_cm_102// -----------private LongAdder curThreadNum = new LongAdder();private long lastFetchTime = -1;
}
com.alibaba.csp.sentinel.node.DefaultNode
java">/** 默认节点 */
public class DefaultNode extends StatisticNode {private ResourceWrapper id; // 关联的资源 (标识)private volatile Set<Node> childList = new HashSet<>(); // 子节点集private ClusterNode clusterNode; // 关联的群集节点。
}
com.alibaba.csp.sentinel.node.EntranceNode
java">/** 入口节点 */
public class EntranceNode extends DefaultNode {... // 只是根据所有的子节点进行计算统计结果
}
com.alibaba.csp.sentinel.node.ClusterNode
java">/** 集群节点 */
public class ClusterNode extends StatisticNode {private final String name; // 名称private final int resourceType; // 资源类型 (0: COMMON; 1: WEB; 2: RPC; 3: ApiGateway; 4: DB)private Map<String, StatisticNode> originCountMap = new HashMap<>(); // 保存不同来源的 StatisticNodeprivate final ReentrantLock lock = new ReentrantLock(); // 操作 originCountMap 的 DCL 锁
}
节点-调用链
- 基本委派给度量指标了
度量指标
- 节点的统计都委派给
ArrayMetric
度量指标-类结构
com.alibaba.csp.sentinel.slots.statistic.metric.Metric
java">/** sign_i_100 度量接口 */
public interface Metric extends DebugSupport {long success(); // 获取总成功数。long maxSuccess(); // 获取最大成功次数。long exception(); // 获取异常总数。long block(); // 获取总阻塞次数。long pass(); // 获取总通过数。 不包括 occupiedPass()long rt(); // 获取总响应时间。long minRt(); // 获得最小的 RT。List<MetricNode> details(); // 获取所有资源的聚合指标节点。MetricBucket[] windows(); // 获取原始窗口数组。void addException(int n); // 添加当前异常计数。void addBlock(int n); // 添加当前阻塞数。void addSuccess(int n); // 添加当前完成的计数。...
}
com.alibaba.csp.sentinel.slots.statistic.metric.ArrayMetric
java">/** sign_c_100 度量实现类 */
public class ArrayMetric implements Metric { // 实现 sign_i_100private final LeapArray<MetricBucket> data; // 统计依然向下委派处理，ref: sing_ac_110// sign_cm_101public ArrayMetric(int sampleCount, int intervalInMs) {this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs); // 秒统计的使用此，ref: sign_cm_110}// sign_cm_102public ArrayMetric(int sampleCount, int intervalInMs, boolean enableOccupy) {if (enableOccupy) {this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);} else {this.data = new BucketLeapArray(sampleCount, intervalInMs); // 分统计的使用此，ref: sign_cm_110}}
}
com.alibaba.csp.sentinel.slots.statistic.base.LeapArray
java">/** sing_ac_110 统计指标的基本数据结构 */
public abstract class LeapArray<T> {protected int windowLengthInMs; // 窗口时间跨度protected int sampleCount;protected int intervalInMs;private double intervalInSecond;protected final AtomicReferenceArray<WindowWrap<T>> array; // T: MetricBucket, ref: sign_c_130 | sing_c_140// 更新锁，仅在当前 bucket 已弃用时使用。private final ReentrantLock updateLock = new ReentrantLock();// sign_cm_110public LeapArray(int sampleCount, int intervalInMs) {... // 省略校验 (两参必须大于 0，且能整除)this.windowLengthInMs = intervalInMs / sampleCount;this.intervalInMs = intervalInMs;this.intervalInSecond = intervalInMs / 1000.0;this.sampleCount = sampleCount;this.array = new AtomicReferenceArray<>(sampleCount);}}
com.alibaba.csp.sentinel.slots.statistic.base.WindowWrap
java">/** sign_c_130 一段时间窗口的包装实体类 */
public class WindowWrap<T> {private final long windowLengthInMs; // 单个窗口桶的时间长度（以毫秒为单位）。private long windowStart; // 窗口的开始时间戳（以毫秒为单位）。private T value; // 统计数据。一般为: MetricBucket, ref: sing_c_140public WindowWrap(long windowLengthInMs, long windowStart, T value) {this.windowLengthInMs = windowLengthInMs;this.windowStart = windowStart;this.value = value;}}
com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket
java">/** sing_c_140 度量桶 (一段时间内的度量数据) */
public class MetricBucket {private final LongAdder[] counters; // sign_f_110 各事件的计数器private volatile long minRt; // 记录最小 RT 值。def: 5000public MetricBucket() {MetricEvent[] events = MetricEvent.values(); // ref: sign_ec_140this.counters = new LongAdder[events.length];for (MetricEvent event : events) {counters[event.ordinal()] = new LongAdder(); // 每个事件，一个计数器}initMinRt(); // 使用配置的最大 RT 值初始化}}
com.alibaba.csp.sentinel.slots.statistic.MetricEvent
java">/** sign_ec_140 事件枚举 */
public enum MetricEvent {PASS,BLOCK,EXCEPTION,SUCCESS,RT,/*** 在未来配额中通过（自 1.5.0 起，预先占用）*/OCCUPIED_PASS
}
度量指标-调用链
addPass()
-
添加
PASS计数 -
com.alibaba.csp.sentinel.slots.statistic.metric.ArrayMetric
java"> @Overridepublic void addPass(int count) {WindowWrap<MetricBucket> wrap = data.currentWindow(); // 获取当前时间戳的桶 ref: sign_m_201wrap.value().addPass(count); // 添加 PASS 计数 ref: sign_m_210}
com.alibaba.csp.sentinel.slots.statistic.base.LeapArray
java"> // sign_m_201 获取当前时间戳的桶public WindowWrap<T> currentWindow() {return currentWindow(TimeUtil.currentTimeMillis()); // sign_m_202}// sign_m_202 在提供的时间戳处获取桶public WindowWrap<T> currentWindow(long timeMillis) {... // 省略小于 0 的判断int idx = calculateTimeIdx(timeMillis); // sign_m_203 计算当前索引long windowStart = calculateWindowStart(timeMillis); // sign_m_204 计算当前窗口的开始时间/*** 从数组中获取给定时间的存储桶。** (1) Bucket 不存在，则只需创建一个新的 Bucket 并 CAS 更新为循环数组。* (2) Bucket 是最新的，那么只需返回 Bucket 即可。* (3) Bucket 已弃用，然后重置当前 Bucket。*/while (true) {WindowWrap<T> old = array.get(idx);if (old == null) { // 桶不存在，则创建/*** newEmptyBucket(timeMillis) 为抽象方法，* 两子类实现：直接返回 new MetricBucket()*/WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));if (array.compareAndSet(idx, null, window)) { // CAS 更新return window; // 更新成功，返回创建的 bucket} else {Thread.yield(); // 争用失败，让出时间片等待可用的桶}} else if (windowStart == old.windowStart()) { // 桶是最新的// 表明另一线程刚好调用上面或下面的逻辑（创建出桶并 CAS 更新完或重置完旧桶）return old;} else if (windowStart > old.windowStart()) { // 桶是旧的/*** 使用锁，保证重置和清理是原子操作*/if (updateLock.tryLock()) {try {return resetWindowTo(old, windowStart); // 重置桶（抽象方法），实现参考: sign_m_206} finally {updateLock.unlock();}} else {Thread.yield(); // 争用失败，让出时间片等待可用的桶}} else if (windowStart < old.windowStart()) {// 不应该到这里...return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));}}}// sign_m_203 计算当前索引，以便将时间戳映射到跳跃 (leap) 数组private int calculateTimeIdx(long timeMillis) {long timeId = timeMillis / windowLengthInMs; // 除窗口时间跨度，得窗口主索引return (int)(timeId % array.length()); // 用主索引取模，映射到数组索引}// sign_m_204 计算窗口的开始时间 (向前取整)protected long calculateWindowStart(long timeMillis) {return timeMillis - timeMillis % windowLengthInMs;}
com.alibaba.csp.sentinel.slots.statistic.metric.BucketLeapArray
java"> // sign_m_206 重置相关统计@Overrideprotected WindowWrap<MetricBucket> resetWindowTo(WindowWrap<MetricBucket> w, long startTime) {// Update the start time and reset value.w.resetTo(startTime); // 重置 bucket 封装的开始时间戳 sign_m_207w.value().reset(); // 重置 bucket 各事件计数器 sign_m_208return w;}
com.alibaba.csp.sentinel.slots.statistic.base.WindowWrap
java"> // sign_m_207 重置当前 bucket 的开始时间戳public WindowWrap<T> resetTo(long startTime) {this.windowStart = startTime;return this;}
com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket
java"> // sign_m_208 重置各事件计数器public MetricBucket reset() {for (MetricEvent event : MetricEvent.values()) {counters[event.ordinal()].reset();}initMinRt();return this;}
com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket
java"> // sign_m_210 添加 PASS 计数public void addPass(int n) {add(MetricEvent.PASS, n); // sign_m_211}// sign_m_211 添加指定事件的计数public MetricBucket add(MetricEvent event, long n) {counters[event.ordinal()].add(n); // 根据枚举的序数定位计数器，然后进行累加 ref: sign_f_110return this;}
pass()
-
获取
PASS计数 -
com.alibaba.csp.sentinel.slots.statistic.metric.ArrayMetric
java"> @Overridepublic long pass() {data.currentWindow(); // 设置当前时间对应的桶，ref: sign_m_201long pass = 0;List<MetricBucket> list = data.values(); // 获取当前“有效”桶的集合 ref: sign_m_220for (MetricBucket window : list) {pass += window.pass(); // 对每个度量桶的 PASS 进行累加 ref: sign_m_225}return pass;}
com.alibaba.csp.sentinel.slots.statistic.base.LeapArray
java"> // sign_m_220 获取当前“有效”桶的集合public List<T> values() {return values(TimeUtil.currentTimeMillis()); // sign_m_221}// sign_m_221 获取指定时间戳“有效”桶的集合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);/*** 为空或当前时间大于桶的开始时间 (桶无效)，则不添加* ref: sign_m_222*/if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {continue;}result.add(windowWrap.value()); // 否则添加返回}return result;}// sign_m_222 判断桶是否无效// 桶开始时间必须小于当前时间才算有效，否则算无效 (返回 true)public boolean isWindowDeprecated(long time, WindowWrap<T> windowWrap) {return time - windowWrap.windowStart() > intervalInMs;}
com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket
java"> // sign_m_225 返回 PASS 计数public long pass() {return get(MetricEvent.PASS); // sign_m_226}// sign_m_226 返回指定事件的计数public long get(MetricEvent event) {return counters[event.ordinal()].sum();}
总结
- 使用
数组 + 时间戳实现滑动窗口，算法简单精妙
