系列文章目录
Flink源码分析: 重启策略机制RestartStrategy
Flink源码分析: Flink JDBC Upsert模式实现原理
前言
这周比较忙，没时间写博客，趁着周五抓紧写一篇，保证业绩。努力的打工人
前两天在群里看到一个小伙伴说他在面试时被面试官问了一套Flink的WaterMark源码，被问了简直一脸懵逼，最后惨遭回去等通知吧....想想自从把公司生产线的Flink从1.10升级到1.12版本后对WaterMark的源码也没怎么深究过，主要是升级到新版本后一直在研究计算SQL话的问题，作为一个多年老司机，怎么可以会了SQL忘了代码的事呢，今天咱们就一起学习下WaterMark的源码吧，当然大家在学习前可以看下我之前写的一篇新版本的WaterMark原理：
使用
Flink1.12版本中建议使用assignTimestampsAndWatermarks来抽象定义水印的生成:
dataStream
.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness[String](Duration.ofSeconds(delayTime))其中WatermarkStrategy继承了TimestampAssignerSupplier<T>接口和WatermarkGeneratorSupplier<T>接口。
先看下TimestampAssignerSupplier<T>的定义：
?创建一个类型为TimestampAssigner的对象，那这个对象有什么作用呢？我们再点进去看一下
?这个接口的作用主要是给每条数据分配一个时间戳，所以说当使用水印时继承了TimestampAssignerSupplier会分配一个时间戳。
WatermarkGeneratorSupplier<T>接口的定义：
onEvent: 每条消息都会调用该方法，传入的消息，消息携带的事件时间，和水印发射器。
onPeriodicEmit：每段时间都会调用该方法,入参是WatermarkOutputoutput，用户可以通过这个方法每隔一段时间发送一次水印，当记录数过多时，每条记录都发送一次水印明显不合适，也影响性能，此时可以通过这个方法进行水印的定时发送，而onEvent只记录当前水印而选择不发射出去。该方法的参数配置为env.getConfig().setAutoWatermarkInterval(300L)，入参是毫秒数，表示隔多少毫秒向下游算子发送一次水印。
而WatermarkStrategy中也提供了一些常用的WatermarkGenerator<T>供用户使用，比如
BoundedOutOfOrdernessWatermarks<T>类中就是一个在onEvent中记录水印，通过onPeriodicEmit方法定时向下游发送水印的实现，构造参数maxOutOfOrderness是提供给记录乱序的，运行最大延迟间隔。MaxTimestamp是当前的水印记录。BoundedOutOfOrdernessWatermarks<T>的大致实现如下:
使用方法也十分的简单，提供的是一个静态方法，只需直接调用即可
WatermarkStrategy.<Map<String,Object>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
最后结合项目，新版本的使用方法如下：
dataStream
.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness[String](Duration.ofSeconds(delayTime))
.withTimestampAssigner(new SerializableTimestampAssigner[String] {
override def extractTimestamp(t: String, l: Long): Long = {
val jSONObject = JSON.parseObject(t)
jSONObject.getString("system_time").toLong
}
}))