处理匹配事件

• 基于 PatternStream 直接调用 select()方法，传入一个 PatternSelectFunction 作为参数,PatternSelectFunction 里需要实现一个 select()方法，这个方法每当检测到一组匹配的复杂事件时都会调用一次。它以保存了匹配复杂事件的 Map 作为输入，经自定义转换后得到输出信息返回。这里我们假设之前定义的模式序列中，有为first和second的两个个体模式，于是可以通过这个名称从 Map 中选择提取出对应的事件。注意调用 Map 的 get(key)方法后得到的是一个事件的 List；如果个体模式是单例的，那么 List 中只有一个元素，直接调用get(0)就可以把它取出。当然，如果个体模式是循环的，List 中就有可能有多个元素了，例：

xxxxxxxxxx
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
  }
})

• PatternStream还有一个类似的方法是 flatSelect()，传入的参数是一个

  PatternFlatSelectFunction。从名字上就能看出，这是PatternSelectFunction的“扁平化”版本；

  内部需要实现一个flatSelect()方法，它与之前 select()的不同就在于没有返回值，而是多了一个收集器（Collector）参数 out，通过调用 out.collet()方法就可以实现多次发送输出数据了。

xxxxxxxxxx
patternStream.flatSelect(new PatternFlatSelectFunction[EventLog,String] {
  override def flatSelect(map: util.Map[String, util.List[EventLog]], 
         collector: Collector[String]): Unit = {
    val first = map.get("first").get(0)
    val second = map.get("first").get(1)
    val third = map.get("first").get(2)
    collector.collect(first.userId + " 连续三次登录失败！登录时间：" + first.timestamp + ", " +
      second.timestamp + ", " + third.timestamp)
  }
}).print("账户锁定")

• PatternStream 的 process()方法，传入一个PatternProcessFunction。这看起来就像是我们熟悉的处理函数（process function），它也可以访问一个上下文（Context），进行更多的操作。所以PatternProcessFunction功能更加丰富、调用更加灵活，可以完全覆盖其他接口，也就成为了目前官方推荐的处理方式。事实上，PatternSelectFunction 和PatternFlatSelectFunction在 CEP 内部执行时也会被转换成PatternProcessFunction。

xxxxxxxxxx
patternStream.process(new PatternProcessFunction[EventLog,String] {
  override def processMatch(
       map: util.Map[String, util.List[EventLog]], 
       context: PatternProcessFunction.Context, 
       collector: Collector[String]): Unit = {
    val first = map.get("first").get(0)
    val second = map.get("first").get(1)
    val third = map.get("first").get(2)
    collector.collect(first.userId + " 连续三次登录失败！登录时间：" + first.timestamp + ", " +
      second.timestamp + ", " + third.timestamp)
  }
}).print("账户锁定")