Flink CEP （四）组合模式

Flink CEP （四）组合模式

文章目录

• 初始模式（Initial Pattern）
• 近邻条件（Contiguity Conditions）
• • 严格近邻（Strict Contiguity）
  • 宽松近邻（Relaxed Contiguity）
  • 非确定性宽松近邻（Non-Deterministic Relaxed Contiguity）
• 其他限制条件
• 循环模式中的近邻条件

定义好的个体模式，就可以尝试按一定的顺序把它们连接起来，定义一个完整的复杂事件匹配规则了。这种将多个个体模式组合起来的完整模式，就叫作“组合模式”（Combining Pattern），为了跟个体模式区分有时也叫作“模式序列”（Pattern Sequence）。一个组合模式有以下形式：

Pattern <Event, ?> pattern = Pattern
			.<Event>begin("start")
			.where(...)
			.next("next")
			.where(...)
			.followedBy("follow")
			.where(...)
			...
1
2
3
4
5
6
7
8

初始模式（Initial Pattern）

所有的组合模式，都必须以一个“初始模式”开头；而初始模式必须通过调用 Pattern 的静态方法.begin()来创建。如下所示：

Pattern<Event, ?> start = Pattern.<Event>begin("start");
1

Pattern 有两个泛型参数，第一个就是检测事件的基本类型 Event，跟 begin 指定的类型一致；第二个则是当前模式里事件的子类型，由子类型限制条件指定。我们这里用类型通配符（?）代替，就可以从上下文直接推断了。

近邻条件（Contiguity Conditions）

初始模式之后，就可以按照复杂事件的顺序追加模式，组合成模式序列了。模式之间的组合是通过一些“连接词”方法实现的，这些连接词指明了先后事件之间有着怎样的近邻关系，这就是所谓的“近邻条件”（Contiguity Conditions，也叫“连续性条件”）。

严格近邻（Strict Contiguity）

匹配的事件严格地按顺序一个接一个出现，中间不会有任何其他事件。代码中对应的就是 Pattern 的.next()方法，名称上就能看出来，“下一个”自然就是紧挨着的。

宽松近邻（Relaxed Contiguity）

宽松近邻只关心事件发生的顺序，而放宽了对匹配事件的“距离”要求，也就是说两个匹配的事件之间可以有其他不匹配的事件出现。代码中对应.followedBy()方法，很明显这表示“跟在后面”就可以，不需要紧紧相邻。

非确定性宽松近邻（Non-Deterministic Relaxed Contiguity）

这种近邻关系更加宽松。所谓“非确定性”是指可以重复使用之前已经匹配过的事件；这种近邻条件下匹配到的不同复杂事件，可以以同一个事件作为开始，所以匹配结果一般会比宽松近邻更多，代码中对应.followedByAny()方法。

其他限制条件

除了上面提到的 next()、followedBy()、followedByAny()可以分别表示三种近邻条件，还可以用否定的“连接词”来组合个体模式。主要包括：
1） .notNext()
表示前一个模式匹配到的事件后面，不能紧跟着某种事件。
2） .notFollowedBy()
表示前一个模式匹配到的事件后面，不会出现某种事件。这里需要注意，由于notFollowedBy()是没有严格限定的；流数据不停地到来，我们永远不能保证之后“不会出现某种事件”。所以一个模式序列不能以 notFollowedBy()结尾，这个限定条件主要用来表示“两个事件中间不会出现某种事件”。

另外，Flink CEP 中还可以为模式指定一个时间限制，这是通过调用.within()方法实现的。方法传入一个时间参数，这是模式序列中第一个事件到最后一个事件之间的最大时间间隔，只有在这期间成功匹配的复杂事件才是有效的。一个模式序列中只能有一个时间限制，调用.within()的位置不限；如果多次调用则会以最小的那个时间间隔为准。

下面是模式序列中所有限制条件在代码中的定义：

// 严格近邻条件
Pattern<Event, ?> strict = start.next("middle").where(...);
// 宽松近邻条件
Pattern<Event, ?> relaxed = start.followedBy("middle").where(...);
// 非确定性宽松近邻条件
Pattern<Event, ?> nonDetermin = start.followedByAny("middle").where(...);
// 不能严格近邻条件
Pattern<Event, ?> strictNot = start.notNext("not").where(...);
// 不能宽松近邻条件
Pattern<Event, ?> relaxedNot = start.notFollowedBy("not").where(...);
// 时间限制条件
middle.within(Time.seconds(10));
1
2
3
4
5
6
7
8
9
10
11
12

循环模式中的近邻条件

循环模式虽说是个体模式，却也可以匹配多个事件；那这些事件之间自然也会有近邻关系的讨论。

在循环模式中，近邻关系同样有三种：严格近邻、宽松近邻以及非确定性宽松近邻。对于定义了量词（如 oneOrMore()、times()）的循环模式，默认内部采用的是宽松近邻。。也就是说，当循环匹配多个事件时，它们中间是可以有其他不匹配事件的；相当于用单例模式分别定义、再用 followedBy()连接起来。。这就解释了在快速上手小节的示例代码中，为什么我们检测连续三次登录失败用了三个单例模式来分别定义，而没有直接指定 times(3)：因为我们需要三次登录失败必须是严格连续的，中间不能有登录成功的事件，而 times()默认是宽松近邻关系。

不过把多个同样的单例模式组合在一起，这种方式还是显得有些笨拙了。连续三次登录失败看起来不太复杂，那如果要检测连续 100 次登录失败呢？显然使用 times()是更明智的选择。不过它默认匹配事件之间是宽松近邻关系，我们可以通过调用额外的方法来改变这一点。

1）.consecutive()

为循环模式中的匹配事件增加严格的近邻条件，保证所有匹配事件是严格连续的。也就是说，一旦中间出现了不匹配的事件，当前循环检测就会终止。这起到的效果跟模式序列中的next()一样，需要与循环量词 times()、oneOrMore()配合使用。

 //定义模式，连续三次登陆失败
        Pattern<LoginEvent, ?> pattern = Pattern.<LoginEvent>begin("first")
                .where(new SimpleCondition<LoginEvent>() { // 以第一个登录失败事件开始
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return value.eventType.equals("fail");
                    }
                }).times(3).consecutive();
1
2
3
4
5
6
7
8

这样显得更加简洁；而且即使要扩展到连续 100 次登录失败，也只需要改动一个参数而已。不过这样一来，后续提取匹配事件的方式也会有所不同，我们将在稍后继续解决此问题。

2）.allowCombinations()
除严格近邻外，也可以为循环模式中的事件指定非确定性宽松近邻条件，表示可以重复使用 已 经 匹 配 的 事 件 。 这 需 要 调 用 .allowCombinations() 方 法 来 实 现 ， 实 现 的 效 果与.followedByAny()相同。