归档
- GitHub: Ali-Sentinel-节点与度量
作用
- 保存资源的实时统计信息
节点
节点-类结构
com.alibaba.csp.sentinel.slots.statistic.metric.DebugSupport
/** 调试支持 */
public interface DebugSupport {
void debug(); // 打印统计信息
}
com.alibaba.csp.sentinel.node.OccupySupport
/** 占用支持 */
public interface OccupySupport {
...
}
com.alibaba.csp.sentinel.node.Node
/** 节点 (用于统计) */
public interface Node extends OccupySupport, DebugSupport {
long totalRequest(); // 获取每分钟传入的请求
long totalPass(); // 获取每分钟的通过次数
long totalSuccess(); // 每分钟完成的请求(Entry.exit())总数
long blockRequest(); // 获取每分钟阻止的请求计数
long totalException(); // 获取每分钟的异常计数
double passQps(); // 获取每秒已通过请求的 QPS
double blockQps(); // 获取每秒被阻止请求的 QPS
double totalQps(); // 获取每秒总请求的 QPS
double successQps(); // 获取每秒已完成请求的(Entry.exit()) QPS
double maxSuccessQps(); // 获得最大已完成请求的 QPS
double exceptionQps(); // 发生异常的 QPS
double 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
/** 统计节点 */
public class StatisticNode implements Node {
// 保存最近 INTERVAL(1000) 毫秒的统计信息。
// 节点的统计逻辑都委派给其处理，ref: sign_i_100 & sign_c_100
private transient volatile Metric rollingCounterInSecond = new ArrayMetric( // sign_cm_101
SampleCountProperty.SAMPLE_COUNT, // def: 2
IntervalProperty.INTERVAL // def: 1000
);
// 保存最近 60 秒的统计信息。
// 节点的统计逻辑都委派给其处理，ref: sign_i_100 & sign_c_100
private 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
/** 默认节点 */
public class DefaultNode extends StatisticNode {
private ResourceWrapper id; // 关联的资源 (标识)
private volatile Set<Node> childList = new HashSet<>(); // 子节点集
private ClusterNode clusterNode; // 关联的群集节点。
}
com.alibaba.csp.sentinel.node.EntranceNode
/** 入口节点 */
public class EntranceNode extends DefaultNode {
... // 只是根据所有的子节点进行计算统计结果
}
com.alibaba.csp.sentinel.node.ClusterNode
/** 集群节点 */
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<>(); // 保存不同来源的 StatisticNode
private final ReentrantLock lock = new ReentrantLock(); // 操作 originCountMap 的 DCL 锁
}
节点-调用链
- 基本委派给度量指标了
度量指标
- 节点的统计都委派给
ArrayMetric
度量指标-类结构
com.alibaba.csp.sentinel.slots.statistic.metric.Metric
/** 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
/** sign_c_100 度量实现类 */
public class ArrayMetric implements Metric { // 实现 sign_i_100
private final LeapArray<MetricBucket> data; // 统计依然向下委派处理，ref: sing_ac_110
// sign_cm_101
public ArrayMetric(int sampleCount, int intervalInMs) {
this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs); // 秒统计的使用此，ref: sign_cm_110
}
// sign_cm_102
public 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
/** 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_110
public 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
/** sign_c_130 一段时间窗口的包装实体类 */
public class WindowWrap<T> {
private final long windowLengthInMs; // 单个窗口桶的时间长度（以毫秒为单位）。
private long windowStart; // 窗口的开始时间戳（以毫秒为单位）。
private T value; // 统计数据。一般为: MetricBucket, ref: sing_c_140
public WindowWrap(long windowLengthInMs, long windowStart, T value) {
this.windowLengthInMs = windowLengthInMs;
this.windowStart = windowStart;
this.value = value;
}
}
com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket
/** sing_c_140 度量桶 (一段时间内的度量数据) */
public class MetricBucket {
private final LongAdder[] counters; // sign_f_110 各事件的计数器
private volatile long minRt; // 记录最小 RT 值。def: 5000
public MetricBucket() {
MetricEvent[] events = MetricEvent.values(); // ref: sign_ec_140
this.counters = new LongAdder[events.length];
for (MetricEvent event : events) {
counters[event.ordinal()] = new LongAdder(); // 每个事件，一个计数器
}
initMinRt(); // 使用配置的最大 RT 值初始化
}
}
com.alibaba.csp.sentinel.slots.statistic.MetricEvent
/** 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
@Override
public void addPass(int count) {
WindowWrap<MetricBucket> wrap = data.currentWindow(); // 获取当前时间戳的桶 ref: sign_m_201
wrap.value().addPass(count); // 添加 PASS 计数 ref: sign_m_210
}
com.alibaba.csp.sentinel.slots.statistic.base.LeapArray
// 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
// sign_m_206 重置相关统计
@Override
protected WindowWrap<MetricBucket> resetWindowTo(WindowWrap<MetricBucket> w, long startTime) {
// Update the start time and reset value.
w.resetTo(startTime); // 重置 bucket 封装的开始时间戳 sign_m_207
w.value().reset(); // 重置 bucket 各事件计数器 sign_m_208
return w;
}
com.alibaba.csp.sentinel.slots.statistic.base.WindowWrap
// sign_m_207 重置当前 bucket 的开始时间戳
public WindowWrap<T> resetTo(long startTime) {
this.windowStart = startTime;
return this;
}
com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket
// 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
// 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_110
return this;
}
pass()
-
获取
PASS计数 -
com.alibaba.csp.sentinel.slots.statistic.metric.ArrayMetric
@Override
public long pass() {
data.currentWindow(); // 设置当前时间对应的桶，ref: sign_m_201
long pass = 0;
List<MetricBucket> list = data.values(); // 获取当前“有效”桶的集合 ref: sign_m_220
for (MetricBucket window : list) {
pass += window.pass(); // 对每个度量桶的 PASS 进行累加 ref: sign_m_225
}
return pass;
}
com.alibaba.csp.sentinel.slots.statistic.base.LeapArray
// 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
// 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();
}
总结
- 使用
数组 + 时间戳实现滑动窗口，算法简单精妙
