关于FlinkCEP的相关概念和说明网上已经有很多介绍的文章，这里就不再赘述，本文主要通过一个简单的场景作为FlinkCEP的入门快速上手，并通过样例初步了解一下Combining Patterns中的事件之间的邻接模式：

FlinkCEP supports the following forms of contiguity between events:

Strict Contiguity: Expects all matching events to appear strictly one after the other, without any non-matching events in-between.

Relaxed Contiguity: Ignores non-matching events appearing in-between the matching ones.

Non-Deterministic Relaxed Contiguity: Further relaxes contiguity, allowing additional matches that ignore some matching events.

To apply them between consecutive patterns, you can use:

next(), for strict,

followedBy(), for relaxed, and

followedByAny(), for non-deterministic relaxed contiguity.

业务场景

对于异常交易产生告警，例如交易分为有效和无效两种，如果先产生了一笔有效交易额小于10，然后产生了一笔有效交易额大于100，就要触发告警。这里为了简化逻辑并没有考虑两笔交易的时间间隔。

业务分析

对于这个业务场景，主要问题在于这两笔交易的连续性，也就是说有三种情况：

两笔交易一定是连续的，且中间无任何的交易产生，也就是两个条件之间用next()连接
两笔交易可以不连续，中间可以有其他的交易，但是最终第二个条件只会匹配上一次成功匹配之后的事件，即会抛弃匹配成功的事件，也就是两个条件之间用followedBy()连接
两笔交易可以不连续，中间可以有其他的交易，并且最终第二个条件会匹配所有满足第一个条件的交易，也就是两个条件之间用followedByAny()连接

第一点很好理解，对于第二，第三点会在稍后程序中做详细的解释。

业务实现

交易抽象为Event.java，其他部分请参见源码

public class Event {
    private EventType type; //事件类型，即有效，无效
    private double volume; //交易额
    private String id; //交易流水号
}

样例CEPExample.java

public class CEPExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "host-10-1-236-139:6667");
        properties.setProperty("group.id", "cepG");
        DataStream<String> stream = env
                .addSource(new FlinkKafkaConsumer010<>("foo", new SimpleStringSchema(), properties));

        DataStream<Event> input = stream.map(new MapFunction<String, Event>() {
            @Override
            public Event map(String value) throws Exception {
                String[] v = value.split(",");
                return new Event(v[0], EventType.valueOf(v[1]), Double.parseDouble(v[2]));
            }
        });

        Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(
                new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        System.out.println(event + " from start");
                        return event.getType() == EventType.VALID && event.getVolume() < 10;
                    }
                }
        ).next("end").where(
                new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        System.out.println(event + " from end");
                        return event.getType() == EventType.VALID && event.getVolume() > 100;
                    }
                }
        );

        PatternStream<Event> patternStream = CEP.pattern(input, pattern);

        DataStream<Alert> result = patternStream.process(
                new PatternProcessFunction<Event, Alert>() {
                    @Override
                    public void processMatch(
                            Map<String, List<Event>> pattern,
                            Context ctx,
                            Collector<Alert> out) throws Exception {

                        System.out.println(pattern);

                        out.collect(new Alert("111", "CRITICAL"));
                    }
                });

        result.print();

        env.execute("Flink cep example");
    }
}

测试数据

id	eventType	volume
1	VALID	2
2	VALID	200
3	VALID	3
4	INVALID	1
5	VALID	1
6	VALID	300
7	VALID	600

结果分析

如果使用的是next("end")，只会触发2次告警，分别为

next("end")

这就是因为next必须要满足两个连续的事件都符合条件。

如果使用的是followedBy("end")，会触发3次告警，分别为

followedBy("end")

可以看到满足条件的event中间可以有不满足的事件产生。

如果使用的是followedByAny("end")，会触发7次告警，分别为

followedByAny("end")

followedByAny("end")和followedBy("end")主要的区别就是所有满足条件的两个事件都会触发告警，即便前一个条件已经生效过。

总结

本文实现了一个简单的CEP场景，并分析了两个事件常见的邻接模式，目前只是初步的一个了解，后续会根据遇到的实际场景再介绍相关的使用方法和原理。

Hello FlinkCEP