- [一、组合模式]
- [1. 初始模式（Initial Pattern）]
- [2. 近邻条件（Contiguity Conditions）]
- [3. 其他限制条件]
- [4.循环模式中的近邻条件]
- [1. .consecutive()]
- [2. .allowCombinations()]
- [二、模式组]
- [三、匹配后跳过策略]
如果个体模式更像编程语言的运算符，那么本篇就是表达式
一、组合模式
有了定义好的个体模式，就可以尝试按一定的顺序把它们连接起来，定义一个完整的复杂事件匹配规则了。这种将多个个体模式组合起来的完整模式，就叫作"组合模式"（Combining Pattern），为了跟个体模式区分有时也叫作"模式序列"（Pattern Sequence）。一个组合模式有以下形式：
Pattern<Event, ?> pattern = Pattern
.<Event>begin("start").where(...)
.next("next").where(...)
.followedBy("follow").where(...)
...
可以看到，组合模式确实就是一个"模式序列"，是用诸如 begin、next、followedBy 等表示先后顺序的"连接词"将个体模式串连起来得到的。在这样的语法调用中，每个事件匹配的条件是什么、各个事件之间谁先谁后、近邻关系如何都定义得一目了然。每一个"连接词"方法调用之后，得到的都仍然是一个 Pattern 的对象；所以从 Java 对象的角度看，组合模式与个体模式是一样的，都是 Pattern。
1. 初始模式（Initial Pattern）
所有的组合模式，都必须以一个"初始模式"开头；而初始模式必须通过调用 Pattern 的静态方法.begin()来创建。如下所示：
Pattern<Event, ?> start = Pattern.<Event>begin("start");
这里我们调用 Pattern 的.begin()方法创建了一个初始模式。传入的 String 类型的参数就是模式的名称；而 begin 方法需要传入一个类型参数，这就是模式要检测流中事件的基本类型， 这里我们定义为 Event。调用的结果返回一个 Pattern 的对象实例。Pattern 有两个泛型参数，第一个就是检测事件的基本类型 Event，跟 begin 指定的类型一致；第二个则是当前模式里事件的子类型，由子类型限制条件指定。我们这里用类型通配符（？）代替，就可以从上下文直接推断了。
2. 近邻条件（Contiguity Conditions）
在初始模式之后，我们就可以按照复杂事件的顺序追加模式，组合成模式序列了。模式之间的组合是通过一些"连接词"方法实现的，这些连接词指明了先后事件之间有着怎样的近邻关系，这就是所谓的"近邻条件"（Contiguity Conditions，也叫"连续性条件"）。
Flink CEP 中提供了三种近邻关系：
- 严格近邻（Strict Contiguity）
匹配的事件严格地按顺序一个接一个出现，中间不会有任何其他事件。代码中对应的就是Pattern 的.next()方法，名称上就能看出来，"下一个"自然就是紧挨着的。 - 宽松近邻（Relaxed Contiguity）
宽松近邻只关心事件发生的顺序，而放宽了对匹配事件的"距离"要求， 也就是说两个匹配的事件之间可以有其他不匹配的事件出现。代码中对应.followedBy()方法， 很明显这表示"跟在后面"就可以，不需要紧紧相邻。
3. 非确定性宽松近邻（Non-Deterministic Relaxed Contiguity）
这种近邻关系更加宽松。所谓"非确定性"是指可以重复使用之前已经匹配过的事件；这种近邻条件下匹配到的不同复杂事件，可以以同一个事件作为开始，所以匹配结果一般会比宽松近邻更多。代码中对应.followedByAny()方法。
从图中可以看到，我们定义的模式序列中有两个个体模式：一是"选择圆形事件"，一是"选择三角形事件"；这时它们之间的近邻条件就会导致匹配出的复杂事件有所不同。很明显，严格近邻由于条件苛刻，匹配的事件最少；宽松近邻可以匹配不紧邻的事件，匹配结果会多一些； 而非确定性宽松近邻条件最为宽松，可以匹配到最多的复杂事件。
3. 其他限制条件
除了上面提到的 next()、followedBy()、followedByAny()可以分别表示三种近邻条件，我们还可以用否定的"连接词"来组合个体模式。主要包括：
- .notNext()
表示前一个模式匹配到的事件后面，不能紧跟着某种事件。 - .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));
4.循环模式中的近邻条件
之前我们讨论的都是模式序列中限制条件，主要用来指定前后发生的事件之间的近邻关系。而循环模式虽说是个体模式，却也可以匹配多个事件；那这些事件之间自然也会有近邻关系的 讨论。
在循环模式中，近邻关系同样有三种：严格近邻、宽松近邻以及非确定性宽松近邻。对于定义了量词（如 oneOrMore()、times()）的循环模式，默认内部采用的是宽松近邻。也就是说， 当循环匹配多个事件时，它们中间是可以有其他不匹配事件的；相当于用单例模式分别定义、再用 followedBy()连接起来。这就解释了在简单案例的示例代码中，为什么我们检测连续三次登录失败用了三个单例模式来分别定义，而没有直接指定 times(3)：因为我们需要三次登录失败必须是严格连续的，中间不能有登录成功的事件，而 times()默认是宽松近邻关系。
不过把多个同样的单例模式组合在一起，这种方式还是显得有些笨拙了。连续三次登录失败看起来不太复杂，那如果要检测连续 100 次登录失败呢？显然使用times()是更明智的选择。不过它默认匹配事件之间是宽松近邻关系，我们可以通过调用额外的方法来改变这一点。
- .consecutive()
为循环模式中的匹配事件增加严格的近邻条件，保证所有匹配事件是严格连续的。也就是说，一旦中间出现了不匹配的事件，当前循环检测就会终止。这起到的效果跟模式序列中的next()一样，需要与循环量词 times()、oneOrMore()配合使用。
于是，简单案例中检测连续三次登录失败的代码可以改成：
Pattern<LoginEvent, LoginEvent> pattern = Pattern
// 第一次
.<LoginEvent>begin("first")
.where(new SimpleCondition<LoginEvent>() {
@Override
public boolean filter(LoginEvent value) throws Exception {
return "fail".equals(value.eventType);
}
}).times(3).consecutive();
这样显得更加简洁；而且即使要扩展到连续 100 次登录失败，也只需要改动一个参数而已。
2. .allowCombinations()
除严格近邻外，也可以为循环模式中的事件指定非确定性宽松近邻条件，表示可以重复使用 已 经 匹 配 的 事 件 。 这 需 要 调 用 .allowCombinations() 方 法 来 实 现 ， 实 现 的 效 果与.followedByAny()相同。
二、模式组
一般来说，代码中定义的模式序列，就是我们在业务逻辑中匹配复杂事件的规则。不过在有些非常复杂的场景中，可能需要划分多个"阶段"，每个"阶段"又有一连串的匹配规则。为了应对这样的需求，Flink CEP 允许我们以"嵌套"的方式来定义模式。
之前在模式序列中，我们用 begin()、next()、followedBy()、followedByAny()这样的"连接词"来组合个体模式，这些方法的参数就是一个个体模式的名称；而现在它们可以直接以一个模式序列作为参数，就将模式序列又一次连接组合起来了。这样得到的就是一个"模式组"（Groups of Patterns）。
在模式组中，每一个模式序列就被当作了某一阶段的匹配条件，返回的类型是一个
GroupPattern。而 GroupPattern 本身是 Pattern 的子类；所以个体模式和组合模式能调用的方法， 比如 times()、oneOrMore()、optional()之类的量词，模式组一般也是可以用的。
具体在代码中的应用如下所示：
// 以模式序列作为初始模式
Pattern<Event, ?> start = Pattern.begin( Pattern.<Event>begin("start_start").where(...)
.followedBy("start_middle").where(...)
);
// 在 start 后定义严格近邻的模式序列，并重复匹配两次
Pattern<Event, ?> strict = start.next( Pattern.<Event>begin("next_start").where(...)
.followedBy("next_middle").where(...)
).times(2);
// 在 start 后定义宽松近邻的模式序列，并重复匹配一次或多次
Pattern<Event, ?> relaxed = start.followedBy( Pattern.<Event>begin("followedby_start").where(...)
.followedBy("followedby_middle").where(...)
).oneOrMore();
//在 start 后定义非确定性宽松近邻的模式序列，可以匹配一次，也可以不匹配
Pattern<Event, ?> nonDeterminRelaxed = start.followedByAny( Pattern.<Event>begin("followedbyany_start").where(...)
.followedBy("followedbyany_middle").where(...)
).optional();
三、匹配后跳过策略
在 Flink CEP 中，由于有循环模式和非确定性宽松近邻的存在，同一个事件有可能会重复利用，被分配到不同的匹配结果中。这样会导致匹配结果规模增大，有时会显得非常冗余。当然，非确定性宽松近邻条件，本来就是为了放宽限制、扩充匹配结果而设计的；我们主要是针对循环模式来考虑匹配结果的精简。
之前已经讲过，如果对循环模式增加了.greedy()的限制，那么就会"尽可能多地"匹配事件，这样就可以砍掉那些子集上的匹配了。不过这种方式还是略显简单粗暴，如果我们想要精确控制事件的匹配应该跳过哪些情况，那就需要制定另外的策略了。
在 Flink CEP 中，提供了模式的"匹配后跳过策略"（After Match Skip Strategy），专门用来精准控制循环模式的匹配结果。这个策略可以在Pattern 的初始模式定义中，作为 begin()的第二个参数传入：
Pattern.begin("start", AfterMatchSkipStrategy.noSkip())
.where(...)
...
匹配后跳过策略 AfterMatchSkipStrategy 是一个抽象类，它有多个具体的实现，可以通过调用对应的静态方法来返回对应的策略实例。这里我们配置的是不做跳过处理，这也是默认策略。
下面我们举例来说明不同的跳过策略。例如我们要检测的复杂事件模式为：开始是用户名为 a 的事件（简写为事件 a，下同），可以重复一次或多次；然后跟着一个用户名为 b 的事件，a 事件和 b 事件之间可以有其他事件（宽松近邻）。用简写形式可以直接写作：“a+ followedByb”。在代码中定义 Pattern 如下：
Pattern.<Event>begin("a").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.user.equals("a");
}
}).oneOrMore()
.followedBy("b").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.user.equals("b");
}
});
我们如果输入事件序列"a a a b"------这里为了区分前后不同的 a 事件，可以记作"a1 a2 a3 b"------那么应该检测到 6 个匹配结果：（a1 a2 a3 b），（a1 a2 b），（a1 b），（a2 a3 b），（a2 b），（a3 b）。如果在初始模式的量词.oneOrMore()后加上.greedy()定义为贪心匹配，那么结果就是：
（a1 a2 a3 b），（a2 a3 b），（a3 b），每个事件作为开头只会出现一次。接下来我们讨论不同跳过策略对匹配结果的影响：
- 不跳过（NO_SKIP）
代码调用AfterMatchSkipStrategy.noSkip()。这是默认策略，所有可能的匹配都会输出。所以这里会输出完整的 6 个匹配。 - 跳至下一个（SKIP_TO_NEXT）
代码调用 AfterMatchSkipStrategy.skipToNext()。找到一个 a1 开始的最大匹配之后，跳过a1 开始的所有其他匹配，直接从下一个 a2 开始匹配起。当然 a2 也是如此跳过其他匹配。最终得到（a1 a2 a3 b），（a2 a3 b），（a3 b）。可以看到，这种跳过策略跟使用.greedy()效果是相同的。 - 跳过所有子匹配（SKIP_PAST_LAST_EVENT）
代码调用AfterMatchSkipStrategy.skipPastLastEvent()。找到 a1 开始的匹配（a1 a2 a3 b）之后，直接跳过所有 a1 直到 a3 开头的匹配，相当于把这些子匹配都跳过了。最终得到（a1 a2 a3 b），这是最为精简的跳过策略。 - 跳至第一个（SKIP_TO_FIRST[a]）
代码调用AfterMatchSkipStrategy.skipToFirst(“a”)，这里传入一个参数，指明跳至哪个模式的第一个匹配事件。找到 a1 开始的匹配（a1 a2 a3 b）后，跳到以最开始一个 a（也就是 a1） 为开始的匹配，相当于只留下 a1 开始的匹配。最终得到（a1 a2 a3 b），（a1 a2 b），（a1 b）。 - 跳至最后一个（SKIP_TO_LAST[a]）
代码调用AfterMatchSkipStrategy.skipToLast(“a”)，同样传入一个参数，指明跳至哪个模式的最后一个匹配事件。找到 a1 开始的匹配（a1 a2 a3 b）后，跳过所有 a1、a2 开始的匹配，跳到以最后一个 a（也就是 a3）为开始的匹配。最终得到（a1 a2 a3 b），（a3 b）。
文章作者: 雨中散步撒哈拉
