Flink窗口触发器

onElement()添加到每个窗口的元素都会调用此方法。
onEventTime()当注册的事件时间计时器触发时，将调用此方法。
onProcessingTime()当注册的处理时间计时器触发时，将调用此方法。
onMerge()与有状态触发器相关，并在两个触发器对应的窗口合并时合并它们的状态，例如在使用会话窗口时。(目前没使用过，了解不多)
clear()执行删除相应窗口时所需的任何操作。(一般是删除定义的状态、定时器等)

registerXXXTimeTimer：注册（处理/事件）时间定时器；
deleteXXXTimeTimer：删除（处理/事件）时间定时器；
getPartitionedState：从Flink状态存储终端获取状态；

CONTINUE：不作任何处理；
FIRE_AND_PURGE：触发窗口计算并输出结果同时清理并释放窗口（该值只会被清理触发器PurgingTrigger使用）；
FIRE：触发窗口计算并输出结果，但窗口并没有被释放并且数据仍然保留；
PURGE：不触发窗口计算，不输出结果，只清除窗口中的所有数据并释放窗口

EventTimeTrigger：基于事件时间和watermark机制来对窗口进行触发计算。
ProcessingTimeTrigger:基于处理时间触发。
CountTrigger:窗口元素数超过预先给定的限制值的话会触发计算。
PurgingTrigger作为其它trigger的参数，将其转化为一个purging触发器。

// 先去注册一个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.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;
}

@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;
}

@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;
}

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;
}