Flink-CEP之NFA编译器

文章目录

$$NFA ^ b$$编译器的作用是将模式对象编译成NFA或者NFAFactory(用来创建多种NFA对象)。这个编译的过程,需要对模式进行拆分从而构建状态以及根据条件构建状态转换信息,最终根据构建好的状态集合来创建NFA。示意图如下:

编译的核心逻辑见方法compileFactory,它接收一个Pattern对象,在介绍Pattern API时我们已经提及过,Pattern可以通过前置指针形成一个Pattern链,而该方法所接收的对象处于链头(第一个构建的Pattern对象看成是链尾),方法中会对Pattern链进行遍历,具体过程如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
public static <T> NFAFactory<T> compileFactory(
Pattern<T, ?> pattern,
TypeSerializer<T> inputTypeSerializer,
boolean timeoutHandling) {
if (pattern == null) {
//如果模式为null,返回一个NFA工厂的实现,且不传递任何状态,意味着将创建一个空的NFA对象
return new NFAFactoryImpl<T>(inputTypeSerializer, 0,
Collections.<State<T>>emptyList(), timeoutHandling);
} else {
//构建一个Map来存储所有生成的状态
Map<String, State<T>> states = new HashMap<>();
long windowTime;
Pattern<T, ?> succeedingPattern;
State<T> succeedingState;
Pattern<T, ?> currentPattern = pattern;
//构建最终态,并加入到Map中,这里将会从Pattern的尾部向头部进行遍历,所以构建的第一个状态是尾部的最终态
State<T> currentState = new State<>(currentPattern.getName(), State.StateType.Final);
states.put(currentPattern.getName(), currentState);
//提取当前Pattern对象的窗口时间
windowTime = currentPattern.getWindowTime() != null
? currentPattern.getWindowTime().toMilliseconds()
: 0L;
//不断向前遍历(不包含第一个Pattern对象)
while (currentPattern.getPrevious() != null) {
//相关变量交换
succeedingPattern = currentPattern;
succeedingState = currentState;
currentPattern = currentPattern.getPrevious();
//获得窗口时间
Time currentWindowTime = currentPattern.getWindowTime();
//如果当前Pattern的窗口时间比其之前Pattern的窗口时间小,则将之前Pattern的窗口时间更新为新的窗口时间
if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
windowTime = currentWindowTime.toMilliseconds();
}
//获取或构建状态
if (states.containsKey(currentPattern.getName())) {
currentState = states.get(currentPattern.getName());
} else {
currentState = new State<>(currentPattern.getName(), State.StateType.Normal);
states.put(currentState.getName(), currentState);
}
//为当前状态设置跟后一个状态之间的转换(边),注意状态转换是“TAKE”,这里同时传入了Pattern所注入的条件
currentState.addStateTransition(new StateTransition<T>(
StateTransitionAction.TAKE,
succeedingState,
(FilterFunction<T>) succeedingPattern.getFilterFunction()));
//如果后一个模式是非紧邻模式,则为当前状态构建自循环的“IGNORE”转换
if (succeedingPattern instanceof FollowedByPattern) {
currentState.addStateTransition(new StateTransition<T>(
StateTransitionAction.IGNORE,
currentState,
null
));
}
}
final State<T> beginningState;
//获取或构建起始状态
if (states.containsKey(BEGINNING_STATE_NAME)) {
beginningState = states.get(BEGINNING_STATE_NAME);
} else {
beginningState = new State<>(BEGINNING_STATE_NAME, State.StateType.Start);
states.put(BEGINNING_STATE_NAME, beginningState);
}
//添加状态转换(起始状态,只能通过“TAKE”向下一状态转换)
beginningState.addStateTransition(new StateTransition<T>(
StateTransitionAction.TAKE,
currentState,
(FilterFunction<T>) currentPattern.getFilterFunction()
));
//以所有的状态构建NFAFactoryImpl对象,它将用来创建NFA对象
return new NFAFactoryImpl<T>(inputTypeSerializer, windowTime,
new HashSet<>(states.values()), timeoutHandling);
}
}

代码段中的NFAFactory用于创建NFA的实例对象,NFAFactoryImpl是其唯一的实现,创建过程中,会将上面收集到的状态集合加入到NFA对象中去。


微信扫码关注公众号:Apache_Flink

apache_flink_weichat


QQ扫码关注QQ群:Apache Flink学习交流群(123414680)

qrcode_for_apache_flink_qq_group