如何停止Apache Flink CEP模式？ 哪个更好使用：KeyedProcessFunction带有状态，计时器，if-else逻辑块或Flink CEP模式库？ 如何停止flink CEP模式？

如何解决如何停止Apache Flink CEP模式？ 哪个更好使用：KeyedProcessFunction带有状态，计时器，if-else逻辑块或Flink CEP模式库？ 如何停止flink CEP模式？

请帮助我，我有两个问题：
我从Apache Kafka json消息中阅读，（然后我有步骤：反序列化为POJO，filter，keyBy ....）

1. 哪个更好使用：KeyedProcessFunction（带有状态，计时器，if-else逻辑块）或Flink CEP模式库？

我可以在 KeyedProcessFunction 中检查输入顺序（检查状态，if-else块，out.collect（...），state.clear（）...您会理解我），如下以及我可以将 Flink CEP库与条件和量化器一起使用。

1. 如何停止flink CEP模式？

例如：
我有输入序列：A1，（无事件1分钟）A2，（无事件5分钟）А3，（无事件1分钟）А4，（无事件5分钟以上）A5。 （在A1和A5之间可能有很多事件）
我想发送输出：A1，A3，A5。
第一个事件，如果下一个事件在上一个事件之后不到5分钟出现，它将不会发送到输出；如果下一个事件在上一个事件之后超过5分钟出现，它将发送到输出。 我应该添加什么图案？？？

Pattern<Event,?> pattern = Pattern.
<Event>begin("start")
.where(new SimpleCondition<Event>(){
 public boolean filter(Event event){
return event.getName().contains("A");
}
}).within(Time.minutes(5));

解决方法

乍看之下，将这个特定示例实现为KeyedProcessFunction似乎微不足道，但是，如果消息可以无序到达，则肯定会出现一些复杂性。然后您可能会愚蠢地认为可能存在实质性差距，而实际上却没有。

但是，如果您想要一个简单的，现成的，现成的解决方案，则此示例非常适合会话窗口。

对于CEP，我认为一个可行的解决方案将具有这种味道：您正在寻找一个A序列（称为A1），紧接着是另一个A（称为A2），其中（A2.timestamp-A1.timestamp ）> = 5分钟。找到匹配项后，发出A1并推进匹配引擎，以便A2成为新的A1。 （方便地，CEP对输入流进行了预排序，因此您不必担心事情会乱七八糟。）