窗口触发器概念
窗口的触发器定义了窗口是何时被触发并同时决定触发行为（对窗口进行清理或者计算）。触发器确定窗口(由窗口分配程序形成)何时准备由窗口函数处理。每个WindowAssigner都带有一个默认触发器。
注意：窗口的触发在内部是设置定时器来实现的。
触发器相关类
triggers包
触发器相关类
Trigger抽象类
Trigger类
触发器接口有五种方法，允许触发器对不同的事件作出反应
onElement()添加到每个窗口的元素都会调用此方法。
onEventTime()当注册的事件时间计时器触发时，将调用此方法。
onProcessingTime()当注册的处理时间计时器触发时，将调用此方法。
onMerge()与有状态触发器相关，并在两个触发器对应的窗口合并时合并它们的状态，例如在使用会话窗口时。(目前没使用过，了解不多)
clear()执行删除相应窗口时所需的任何操作。(一般是删除定义的状态、定时器等)
TriggerContext接口（定义在Trigger类中），用于维持状态，注册定时器等：
TriggerContext.png
registerXXXTimeTimer：注册（处理/事件）时间定时器；
deleteXXXTimeTimer：删除（处理/事件）时间定时器；
getPartitionedState：从Flink状态存储终端获取状态；
TriggerResult枚举类，用于决定窗口在触发后的行为：
image.png
CONTINUE：不作任何处理；
FIRE_AND_PURGE：触发窗口计算并输出结果同时清理并释放窗口（该值只会被清理触发器PurgingTrigger使用）；
FIRE：触发窗口计算并输出结果，但窗口并没有被释放并且数据仍然保留；
PURGE：不触发窗口计算，不输出结果，只清除窗口中的所有数据并释放窗口
Flink内置的触发器:
EventTimeTrigger：基于事件时间和watermark机制来对窗口进行触发计算。
ProcessingTimeTrigger:基于处理时间触发。
CountTrigger:窗口元素数超过预先给定的限制值的话会触发计算。
PurgingTrigger作为其它trigger的参数，将其转化为一个purging触发器。
源码剖析
Trigger有以下主要方法：
// 先去注册一个ProcessingTime定时器，触发时间点就是当前窗口的最大时间戳；
//触发结果就是不做任何操作。
public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;
// ProcessingTime到来之时调用
public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;
// EventTime到来之时调用
public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;
- EventTimeTrigger
当窗口的最大时间戳小于等于水位线立即触发窗口计算；否则的话就去注册EventTime定时器，结果就是不做任何操作。
// EventTimeTrigger.java
// 针对每个元素触发
@Override
public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
// if the watermark is already past the window fire immediately
// window的最大时间戳比watermark小，该window需要立刻进行计算
return TriggerResult.FIRE;
} else {
// 注册一个event time事件，当watermark超过window.maxTimestamp时，会调用onEventTime方法
ctx.registerEventTimeTimer(window.maxTimestamp());
return TriggerResult.CONTINUE;
}
}
@Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
// 当时间等于窗口最大时间戳，则触发对窗口进行计算，否则不做任何操作
return time == window.maxTimestamp() ?
TriggerResult.FIRE :
TriggerResult.CONTINUE;
}
@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
// 对于processing time，不做任何处理
return TriggerResult.CONTINUE;
}
- ProcessingTimeTrigger
依照ProcessingTime判断是否触发计算。源码如下：
@Override
public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
// 和eventtime的类似，注册processing time定时器
ctx.registerProcessingTimeTimer(window.maxTimestamp());
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
// 返回FIRE，执行计算
return TriggerResult.FIRE;
}
- CountTrigger
每maxCount个元素触发一次计算。
@Override
public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
// 获取目前的count状态值
ReducingState<Long> count = ctx.getPartitionedState(stateDesc);
count.add(1L);
// 如果count值大于等于maxCount，清空状态值并且触发计算，否则不进行操作
if (count.get() >= maxCount) {
count.clear();
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
// 不进行操作
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
// 不进行操作
return TriggerResult.CONTINUE;
}
PurgingTrigger
PurgingTrigger是一个trigger的包装类。具体作用为：如果被包装的trigger触发返回FIRE，则PurgingTrigger将返回修改为FIRE_AND_PURGE，其他的返回值不做处理。DeltaTrigger
DeltaTrigger具有一个DeltaFunction，该函数的逻辑需要用户自己定义。该函数比较上一次触发计算的元素，和目前到来的元素。比较结果为一个double类型阈值。如果阈值超过DeltaTrigger配置的阈值，会返回TriggerResult.FIRE
public TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception {
// 获取上个元素的储存状态
ValueState<T> lastElementState = ctx.getPartitionedState(stateDesc);
// 确保存入第一个到来的元素
if (lastElementState.value() == null) {
lastElementState.update(element);
return TriggerResult.CONTINUE;
}
if (deltaFunction.getDelta(lastElementState.value(), element) > this.threshold) {
// 重点：只要触发条件满足的时候，才会更新lastElementState，使用新的element替代上一个element
lastElementState.update(element);
// 触发计算
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
