上一篇我们对Flink CEP做了简单介绍，这一篇我们通过代码来演示一下Flink CEP SQL中的严格近邻效果：

（1）pom依赖：

    org.apache.flink    flink-cep_${scala.binary.version}    ${flink.version}

（2）定义一个消息对象

public static class Ticker {    public long id;    public String symbol;    public long price;    public long tax;    public LocalDateTime rowtime;    public Ticker() {    }    public Ticker(long id, String symbol, long price, long item, LocalDateTime rowtime) {        this.id = id;        this.symbol = symbol;        this.price = price;        this.tax = tax;        this.rowtime = rowtime;    }}

（3）构造数据，定义事件组合

public static void main(String[] args) {    EnvironmentSettings settings = null;    StreamTableEnvironment tEnv = null;    try {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        settings = EnvironmentSettings.newInstance()                .useBlinkPlanner()                .inStreamingMode()                .build();        tEnv = StreamTableEnvironment.create(env, settings);        System.out.println("===============CEP_SQL_9=================");        final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");        DataStream dataStream =            env.fromElements(                new Ticker(1, "ACME", 22, 1, LocalDateTime.parse("2021-12-10 10:00:00", dateTimeFormatter)),                new Ticker(3, "ACME", 19, 1, LocalDateTime.parse("2021-12-10 10:00:02", dateTimeFormatter)),                new Ticker(4, "ACME", 23, 3, LocalDateTime.parse("2021-12-10 10:00:03", dateTimeFormatter)),                new Ticker(5, "Apple", 25, 2, LocalDateTime.parse("2021-12-10 10:00:04", dateTimeFormatter)),                new Ticker(6, "Apple", 18, 1, LocalDateTime.parse("2021-12-10 10:00:05", dateTimeFormatter)),                new Ticker(7, "Apple", 16, 1, LocalDateTime.parse("2021-12-10 10:00:06", dateTimeFormatter)),                new Ticker(8, "Apple", 14, 2, LocalDateTime.parse("2021-12-10 10:00:07", dateTimeFormatter)),                new Ticker(9, "Apple", 15, 2, LocalDateTime.parse("2021-12-10 10:00:08", dateTimeFormatter)),                new Ticker(10, "Apple", 25, 2, LocalDateTime.parse("2021-12-10 10:00:09", dateTimeFormatter)),                new Ticker(11, "Apple", 22, 1, LocalDateTime.parse("2021-12-10 10:00:11", dateTimeFormatter)),                new Ticker(12, "Apple", 15, 1, LocalDateTime.parse("2021-12-10 10:00:12", dateTimeFormatter)),                new Ticker(13, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:13", dateTimeFormatter)),                new Ticker(14, "Apple", 25, 1, LocalDateTime.parse("2021-12-10 10:00:14", dateTimeFormatter)),                new Ticker(15, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:15", dateTimeFormatter)),                new Ticker(16, "Apple", 15, 1, LocalDateTime.parse("2021-12-10 10:00:16", dateTimeFormatter)),                new Ticker(17, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:17", dateTimeFormatter)),                new Ticker(18, "Apple", 15, 1, LocalDateTime.parse("2021-12-10 10:00:18", dateTimeFormatter)));        Table table = tEnv.fromDataStream(dataStream, Schema.newBuilder()                .column("id", DataTypes.BIGINT())                .column("symbol", DataTypes.STRING())                .column("price", DataTypes.BIGINT())                .column("tax", DataTypes.BIGINT())                .column("rowtime", DataTypes.TIMESTAMP(3))                .watermark("rowtime", "rowtime - INTERVAL '1' SECOND")                .build());        tEnv.createTemporaryView("CEP_SQL_9", table);        String sql = "SELECT * " +                "FROM CEP_SQL_9 " +                "    MATCH_RECOGNIZE ( " +                "        PARTITION BY symbol " +       //按symbol分区，将相同卡号的数据分到同一个计算节点上。                "        ORDER BY rowtime " +          //在窗口内，对事件时间进行排序。                "        MEASURES " +                   //定义如何根据匹配成功的输入事件构造输出事件                "            e1.id as id,"+                "            AVG(e1.price) as avgPrice,"+                "            e1.rowtime AS start_tstamp, " +                "            e3.rowtime AS end_tstamp " +                "        ONE ROW PER MATCH " +                                      //匹配成功输出一条                "        AFTER MATCH  skip to next row " +                   //匹配后跳转到下一行                "        PATTERN ( e1 e2 e3) WITHIN INTERVAL '2' MINUTE" +                "        DEFINE " +                                                 //定义各事件的匹配条件                "            e1 AS " +                "                e1.price = 25 , " +                "            e2 AS " +                "                e2.price > 10 ," +                "            e3 AS " +                "                e3.price = 15 " +                "    ) MR";                        TableResult res = tEnv.executeSql(sql);        res.print();        tEnv.dropTemporaryView("CEP_SQL_9");            } catch (Exception e) {                LOG.error(e.getMessage(), e);            }}

（4）关键代码解释：

输出两分钟内匹配到的数据，输出信息：

（5）执行效果：

从数据集中匹配到了两组符合要求的数据。