Flink源码系列——指标监测_flink metrics 查看每秒的tps

1、Metric简介

Flink对于指标监测有一套自己的实现，指标的统计方式有四种，这些指标都实现了Metric这个接口，而Metric这个接口只是一个标识，本身并没有定义如何方法接口，部分子类的继承关系如下所示。

从图中可以看出，Metric这个接口有四个直接子类，分别是：

Gauge —— 最简单的度量指标，只是简单的返回一个值，比如返回一个队列中当前元素的个数；
Counter —— 计数器，在一些情况下，会比Gauge高效，比如通过一个AtomicLong变量来统计一个队列的长度；
Meter —— 吞吐量的度量，也就是一系列事件发生的速率，例如TPS；
Histogram —— 度量值的统计结果，如最大值、最小值、平均值，以及分布情况等。

以MeterView为例，分析一个Metric的具体实现。MeterView还实现View接口，实现View接口的类，表示其会定期的执行update方法，进行数据的更新。

public class MeterView implements Meter, View {
   
   /** 底层使用的计算器 */
   private final Counter counter;
   /** 计算平均值的事件跨度 */
   private final int timeSpanInSeconds;
   /** 包含历史数据的循环数组 */
   private final long[] values;
   /** 当前时间在数组中的索引 */
   private int time = 0;
   /** 最新计算的rate */
   private double currentRate = 0;

   public MeterView(int timeSpanInSeconds) {
      this(new SimpleCounter(), timeSpanInSeconds);
   }

   public MeterView(Counter counter, int timeSpanInSeconds) {
      this.counter = counter;
      /** 这里的操作是为了让时间跨度刚好是 UPDATE_INTERVAL_SECONDS 的整数倍 */
      this.timeSpanInSeconds = timeSpanInSeconds - (timeSpanInSeconds % UPDATE_INTERVAL_SECONDS);
      this.values = new long[this.timeSpanInSeconds / UPDATE_INTERVAL_SECONDS + 1];
   }

   @Override
   public void markEvent() {
      this.counter.inc();
   }

   @Override
   public void markEvent(long n) {
      this.counter.inc(n);
   }

   @Override
   public long getCount() {
      return counter.getCount();
   }

   @Override
   public double getRate() {
      return currentRate;
   }

   @Override
   public void update() {
      time = (time + 1) % values.length;
      values[time] = counter.getCount();
      currentRate =  ((double) (values[time] - values[(time + 1) % values.length]) / timeSpanInSeconds);
   }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

从类的属性变量中可以看出，MeterView是在一个Counter计数器的基础之上，封装了一层，从而实现事件每秒的平均速率。以values这个长整型的数组，作为环形数组，实现对最新的历史数据的保存。
在构造函数中，会对入参timeSpanInSeconds这个时间跨度进行修正，使其刚好是UPDATE_INTERVAL_SECONDS的整数倍，另外values数组的长度是timeSpanInSeconds对UPDATE_INTERVAL_SECONDS倍数，再加上1，这样这个数组的最新数据和最老的数据之间的时间间隔就刚好是timeSpanInSeconds。
假设values数组的长度为n，则：

1、索引n-1处的统计值，和索引0处的统计值，时间间隔就是timeSpanInSeconds；
2、由于是环形数组，所以索引0处的统计值，和索引1处的统计值的时间间隔就是timeSpanInSeconds；
3、所以索引i处的统计值，和索引(i+1)%n处的统计值，时间间隔是timeSpanInSeconds；

这个逻辑理清楚了，对update方法的逻辑也就清楚了。

另外，对于Metrics相关概念，可以参考 http://wuchong.me/blog/2015/08/01/getting-started-with-metrics/

---

2、MetricGroup

为了便于对Metric进行方便的管理和区分，可以对Metric进行分组，MetricGroup就是用来实现这个功能的。
MetricGroup的相关子类的继承关系如下所示。

1、ProxyMetricGroup —— 这是一个代理类，就是把新Metric或者新的子MetricGroup的注册，委托给代理MetricGroup进行处理；
2、AbstractMetricGroup —— 对新增Metric和子MetricGroup的相关方法进行了实现；

在AbstractMetricGroup中有这些属性

protected final A parent;
private final Map<String, Metric> metrics = new HashMap<>();
private final Map<String, AbstractMetricGroup> groups = new HashMap<>();
1
2
3

parent —— 用来保存这个MetricGroup的父MetricGroup
metrics —— 这个map，是用来保存当前MetricGroup中注册的Metric；
groups —— 这个map，是用来保存当前MetricGroup中注册子MetricGroup；

通过这个数据结构可以看出，在MetricGroup中，可以建立一个树状的结构，用来存储和归类相关的Metric。

---

3、MetricReporter

MetricReporter是用来向外披露Metric的监测结果的接口。
由于MetricReporter的子类在实例化时，都是通过反射机制，所以对于其实现子类，需要有一个公共，无参的构造函数，这个接口的定义如下：

public interface MetricReporter {
   void open(MetricConfig config);
   void close();
   void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group);
   void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group);
}
1
2
3
4
5
6