实践指路明灯，源码剖析flink-metrics

1

通过上期的分享，我们对 Metrics 类库有了较深入的认识，并对指标监控的几个度量类型了如指掌。

本期，我们将走进当下最火的流式处理框架 flink 的源码，一同深入并学习一下别人家的代码。

2

会当凌绝顶，一览众山小。先从全局了解个梗概，然后再采用剥洋葱的方式逐层去解密。

我本地的源码是 flink-1.8.1 版本，打开源码，进入 flink-metrics 模块，发现很多以 flink-metrics-开头的系列模块。

映入眼帘的 flink-metrics 系列模块虽然很多，不用发愁。其实主要分为指标监控基础核心模块（flink-metrics-core），以及指标数据监控组件集成模块（flink-metrics-xxx）两大类。

2.1

flink-metrics-core 模块剖析。

开篇提到 Metrics 的几种度量类型，来看看 flink 是咋定义的？

打开指标监控基础核心模块 flink-metrics-core 看个梗概。

不出我们所料，flink 定义了 Metrics 监控中常见的几种度量规范（Meter、Gauge、Counter、Histogram），画个简易的类图再看的明白些。

不过为了便于管理和区分 Metric，于是就有了 MetricGroup 的定义，那么编码时可以直接与 MetricGroup 交互就可以啦。

到这，了解了全局，不妨采用剥洋葱的方式，再去看看每个接口定义的都是啥规范？逐一进行解密。

Meter 定义 getRate() 方法，用于统计系统中某一个事件的速率，定义 getCount() 方法，用于系统中事件的计数统计。

Gauge 是最简单的度量指标，只有一个简单的返回值，定义 getValue() 方法，用来获取一些对象或者事物的瞬时值。

Counter 累计型的度量指标，定义 inc() 累加方法，以及 dec() 累减等方法。

Histogram 是一种非常常见的统计图表，统计数据的分布情况，其中定义的 getStatistics() 方法，提供了最小值，最大值，中间值等对应的计算支撑。

其中 HistogramStatistics 定义如下。

指标对外披露，flink 是咋定义的？

是否还记得上期的分享中，如何把指标数据披露出去的？没错，那就是有一系列的 Reporter 来完成的事情，接下来看看 flink 有没有类似的规范定义呢？

也是没出我们所料，flink 定义了专门用于指标报告的规范，画个简易的类图，从全局上稍微看的明白些。

如上图示意，flink 定义了 MetricReporter 用于规定指标披露规范，其中 AbstractReporter 提供了 MetricReporter 接口的简单的实现。Scheduled 接口用于定义数据披露方案，提供定期性的披露当前数据的支持。

那么，不妨采取剥洋葱的方式，逐个再去看看都是啥，能意会就行啦。

MetricReporter 是 flink 定义的指标披露的接口规范，方法名字其实已经很明确啦，再解说一二。其中 open() 方法主要用于初始化相关操作；close() 主要是关闭 Reporter，用于释放资源；notifyOfAddedMetric()/notifyOfRemovedMetric() 定义新指标注册以及指标删除时通知。

如上图示意，AbstractReporter 抽象类实现了接口 MetricReporter，并拥有 gauges、counters、histograms、meters 四个度量类型的 Map 容器，并对接口 MetricReporter 中定义的 notifyOfAddedMetric()/notifyOfRemovedMetric() 提供了简单实现。

Scheduled 接口很简单，定义了 report() 方法，用于定期披露当前指标数据。

上面对 flink-metrics-core 指标监控核心定义模块，有了初步的认识，那么看看这些规范到底是怎么用的？

2.2

flink-metrics-prometheus 模块剖析。

从全局上了解个梗概，画个简易的类图，稍微看的更清晰。

很显然，针对 Prometheus 做了支撑，没有使用 flink-metrics-core 包中的 AbstractReporter 类，而是对 MetricReporter 进行了单独实现，不妨我们去看个梗概。

见名知意，AbstractPrometheusReporter 是对 Prometheus 轮子单独定义的，实现了 MetricReporter 接口定义的方法，肯定少不了去迎合 Prometheus 的代码，摘取部分核心代码剖析一下。

如上图所示，notifyOfAddedMetric() 方法主要按照 Prometheus 的规范进行封装成 io.prometheus.client.Collector，以及调用 Prometheus 提供的方法对 Metric 进行转换封装（说白了，就是按照 Prometheus 的规范进行转换，不懂也没关系，因为每个轮子的实现均不同）。

再去看 AbstractPrometheusReporter 实现类 PrometheusPushGatewayReporter 的源码之前，不妨先扒拉出以往分享 Prometheus 实战中的一张图，解说一二。

如图示意，要完成数据源层的 AppService 的指标监控，需要对应用纳入一个 Client lib 来支撑指标产生、输出，然后通过 push 的形式，主动推送数据到 PushGateway，然后 Prometheus Server 会定期性的从 PushGateway 上拉取指标数据。

知道了流程，我们再去看 flink-metrics-prometheus 的源码，可能会更清晰。

如源码所示，PrometheusPushGatewayReporter 类持有 PushGateway 的引用，进而可以使用 Prometheus 提供的 API 进行 push 数据，以及创建连接和释放资源。

代码很简单，open() 方法主要读取配置文件，进行实例化 PushGateway；report() 方法会调用 PushGateway 提供的 push 方法进行推送数据；close() 方法主要用于释放资源，从 PushGateway 上把 jobName 对应的 metrics 删除。

如果你一直追随一猿小讲的脚步的话，那么会想起之前 flink 监控实践的一段配置，结合上段代码，再去看这段配置，就知道这个配置咋回事啦。

搞定了监控数据如何披露出去的，唯独有一点，还是没有搞明白？那就是 PrometheusPushGatewayReporter 什么时候实例化的呢？也就是流程怎么串起来的呢？

2.3

flink-runtime 模块中部分代码剖析。

首先找到flink-runtime 目录下的 metrics 包中的 org.apache.flink.runtime.metrics.MetricRegistryImpl 一探究竟。

截图为 MetricRegistryImpl 构造方法的部分代码，其中标注 1 代码，主要是依据配置内容，构建配置的 Reporter 对象；标注 2 的代码，调用其 open() 方法，完成初始化动作；标注 3 的代码，主要是判断 Reporter 对象有没有实现 Scheduled 接口，如果实现了，则会通过 executor 进行定期调用 reporter.report() 方法进行报送数据。

了解到指标数据如何对外披露，但是 Metric 在哪儿添加的呢？

我们知道 MetricGroup 设计的目的，是为了方便对 Metric 管理，而 AbstractMetricGroup 则对添加 Metric 等相关方法进行了定义实现。

如图示标注 2 代码，AbstractMetricGroup 类中会调用 registry.register(metric, name, this) 完成指标的注册添加，进而会调用 MetricRegistryImpl 的 register 方法。

如图中标注 2 代码，会调用对应的 Reporter 的 notifyOfAddedMetric() 方法完成添加指标通知。

此时，代码懵懵懂，再去看以往实战的效果图（Granfna 展示 flink-metrics）的背后，脑海里应该会清晰不少。

3

flink-metrics 源码剖析就到这儿，flink 与 prometheus 监控集成剖析也刨到这儿，其实 flink-metrics 与其它轮子集成也是这么一回事儿，掌握脉略就行了。

另外，flink 作为当下最流行的开源流式处理框架，那么 flink 定义的指标度量接口规范，势必会对我们应用监控有建设性的参考意义。

好了，本次分享就到这儿，希望你们能够喜欢。