Pattern量词

模式（Pattern）其实就是将一组简单事件组合成复杂事件的“匹配规则”。由于流中事件的匹配是有先后顺序的，因此一个匹配规则就可以表达成先后发生的一个个简单事件，按顺序串联组合在一起。

这里的每一个简单事件并不是任意选取的，也需要有一定的条件规则；所以我们就把每个

简单事件的匹配规则，叫作“个体模式”（Individual Pattern）。

• 基本形式

快速入门实战案例中，每一个登录失败事件的选取规则，就都是一个个体模式。比如：

xxxxxxxxxx
.begin[EventLog]("first") 
.where(_.eventType.equals("fail"))

或者后面的：

xxxxxxxxxx
.next("second")
.where(_.eventType.equals("fail"))

这些都是个体模式。个体模式一般都会匹配接收一个事件。

每个个体模式都以一个“连接词”开始定义的，比如 begin、next 等等，这是 Pattern 对象

的一个方法，返回的还是一个 Pattern。这些“连接词”方法有一个 String 类型参数，这就是

当前个体模式唯一的名字，比如这里的“first”、“second”。在之后检测到匹配事件时，就会以这个名字来指代匹配事件。

个体模式需要一个“过滤条件”，用来指定具体的匹配规则。这个条件一般是通过调用

另外，个体模式可以匹配接收一个事件，也可以接收多个事件。这听起来有点奇怪，一个

单独的匹配规则可能匹配到多个事件吗？这是可能的，我们可以给个体模式增加一个“量词”

（quantifier），就能够让它进行循环匹配，接收多个事件。

• 量词（Quantifiers ）

个体模式后面可以跟一个“量词”，用来指定循环的次数。从这个角度分类，个体模式可

以包括“单例（singleton）模式”和“循环（looping）模式”。默认情况下，个体模式是单例

模式，匹配接收一个事件；当定义了量词之后，就变成了循环模式，可以匹配接收多个事件。

在循环模式中，对同样特征的事件可以匹配多次。比如我们定义个体模式为“匹配形状为

三角形的事件”，再让它循环多次，就变成了“匹配连续多个三角形的事件”。注意这里的“连

续”，只要保证前后顺序即可，中间可以有其他事件，所以是“宽松近邻”关系。

在 Flink CEP 中，可以使用不同的方法指定循环模式，主要有：

• oneOrMore

匹配事件出现一次或多次，假设 a 是一个个体模式，a.oneOrMore 表示可以匹配 1 个或多

个 a 的事件组合。我们有时会用 a+来简单表示。

• times：可以通过times指定固定的循环执行次数。

xxxxxxxxxx
/**举例：事件流：a1,b1,a2,b2,a3,a4,a5,a6,a7... ....(a1代表第一次出现a,其他以此类推，a5第5次出现事件a)
*假设start.where(匹配a事件).times(4) ,匹配结果如下：
*  {a1,a2,a3,a4},{a2,a3,a4,a5},{a3,a4,a5,a6}
*假设start.where(匹配a事件).times(2,4) ,匹配结果如下：
*  {a1,a2},{a1,a2,a3},{a2,a3},{a1,a2,a3,a4},{a2,a3,a4},{a3,a4}... ...
*/
//当where条件满足指定次数后，循环触发,连续出现4次匹配条件就触发，中间可以有其他事件
start.times(4);//应出现4次
//可以执行触发次数范围,让循环执行次数在该范围之内，中间可以有其他事件
start.times(2, 4);//应出现2、3、4次

• consecutive()

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

  xxxxxxxxxx
  // 1. 定义 Pattern，登录失败事件，循环检测 3 次
  val pattern = Pattern.begin[EventLog]("first")
        .where(_.eventType.equals("fail"))
        .times(3).consecutive()
  patternStream
    .select(new PatternSelectFunction[EventLog, String] {
      override def select(map: util.Map[String, util.List[EventLog]]): String
      = {
        val first = map.get("first").get(0)
        val second = map.get("first").get(1)
        val third = map.get("first").get(2)
        first.userId + " 连续三次登录失败！登录时间：" + first.timestamp + ", " +
          second.timestamp + ", " + third.timestamp
      }
    })
    .print("账户锁定")
  

• optional：也可以通过optional关键字指定要么不触发要么触发指定的次数。需要与多个模式组合时，才有意义。

xxxxxxxxxx
// expecting 4 occurrences 应出现4次
start.times(4)
// expecting 0 or 4 occurrences应出现0或4次
start.times(4).optional()
// expecting 2, 3 or 4 occurrences应出现2、3或4次
start.times(2, 4)
// expecting 0, 2, 3 or 4 occurrences应出现0、2、3或4次
start.times(2, 4).optional()

• 终止条件：如果程序中使用了oneOrMore或者oneOrMore().optional()方法，则必须指定终止条件，否则模式中的规则会一直循环下去，如下终止条件通过until()方法指定。

  xxxxxxxxxx
  start.oneOrMore.until(_.callOut.startsWith("186"))
  

• greedy可以通过greedy将Pattern标记为贪婪模式，在Pattern匹配成功的前提下，会尽可能多地触发。需要与多个模式组合时才有意义。

xxxxxxxxxx
// expecting 2, 3 or 4 occurrences and repeating as many as possible期望出现2、3或4次，并尽可能多地重复
start.times(2, 4).greedy()
//触发0、2、3、4次,尽可能重复执行
//需要多个条件组合一起使用，才有效果。单独使用没意义
start.times(2, 4).optional.greedy;

比如：

xxxxxxxxxx
val pattern: Pattern[String, String] = Pattern.begin[String]("start").where(s => {
  s.startsWith("a")
}).times(2, 3)
  /**注意：加上greedy:就是尽可能多的往后匹配start事件【这里尽可能多不是说组合多】，与middle满足 pattern
   */
  .greedy
  .next("middle").where(s => {
    s.length == 3
  })