带时间戳的闪烁计数器

Question

我正在阅读Flink示例CountWithTimestamp，下面是该示例的代码片段：

@Override
public void processElement(Tuple2<String, String> value, Context ctx, Collector<Tuple2<String, Long>> out)
throws Exception {
// retrieve the current count
CountWithTimestamp current = state.value();
if (current == null) {
current = new CountWithTimestamp();
current.key = value.f0;
}
// update the state's count
current.count++;
// set the state's timestamp to the record's assigned event time timestamp
current.lastModified = ctx.timestamp();
// write the state back
state.update(current);
// schedule the next timer 60 seconds from the current event time
ctx.timerService().registerEventTimeTimer(current.lastModified + 60000);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out)
throws Exception {
// get the state for the key that scheduled the timer
CountWithTimestamp result = state.value();
// check if this is an outdated timer or the latest timer
if (timestamp == result.lastModified + 60000) {
// emit the state on timeout
out.collect(new Tuple2<String, Long>(result.key, result.count));
}
}
}

我的问题是，如果我删除onTimer中的if语句timestamp == result.lastModified + 60000(collect stmt not touch(，而在processElement的开头用另一个if语句if(ctx.timestamp < current.lastModified + 60000) { deleteEventTimeTimer(current.lastModified + 60000)}替换它，程序的语义会相同吗？在语义相同的情况下，一个版本比另一个版本有什么偏好？

Answer 1

您认为删除计时器的实现具有相同的语义是正确的。事实上，我最近改变了培训材料中使用的例子，因为我更喜欢这种方法。我觉得它更可取的原因是，所有复杂的业务逻辑都在一个地方(在processElement中(，并且无论何时调用onTimer，您都可以准确地知道该做什么，而不会提出任何问题。此外，它的性能更高，因为检查点和最终触发的计时器更少。

这个例子是在定时器被删除之前为文档编写的，还没有更新。

你可以找到我在这些幻灯片中提到的修改过的例子——https://training.ververica.com/decks/process-function/——一旦你通过了注册页面。

FWIW，我最近也按照同样的思路重新制定了相应训练练习的参考解决方案：https://github.com/apache/flink-training/tree/master/long-ride-alerts.