1.业务需求
实时计算需求 :
业务背景：支付宝智能助理上线后，用户只需从支付宝首页轻松下拉即可体验，因此得到了大量用户的使用和反馈。
业务需求：现在业务方想统计每个用户在智能助理近30min、近1h、近6h的会话数和会话时长，并随着时间推移，每1分钟更新一次。
额外诉求：如果用户在30min、1h、6h后没有行为数据，则将会话数和会话时长清零。
2.方案设计
数据流处理：使用KeyedProcessFunction按用户ID分组，维护用户状态。
状态管理：为每个用户维护三个窗口（30min、1h、6h）的统计信息（会话数、会话时长、最后一次事件时间）。
定时器触发：每分钟触发一次定时器，检查用户状态是否过期并清零统计值。
Watermark机制：基于事件时间处理，确保时间窗口的准确性。
3.代码实现
3.1 定义数据类型
// 用户会话事件
public class SessionEvent {
private String userId;
private long eventTime; // 事件时间（毫秒）
private long sessionDuration; // 会话时长（毫秒）
// 构造函数、getter/setter
}
// 用户统计结果
public class UserStats {
private String userId;
private long count30min, duration30min;
private long count1h, duration1h;
private long count6h, duration6h;
// 构造函数、getter/setter
}
3.2 用户状态管理类
public class UserWindowState {
private static final int[] WINDOW_DURATIONS = {30 * 60 * 1000, 60 * 60 * 1000, 6 * 60 * 60 * 1000};
private long[] sessionCounts = new long[3];
private long[] durations = new long[3];
private long lastEventTime;
// 更新指定窗口的统计值
public void update(int windowIndex, long countDelta, long durationDelta) {
sessionCounts[windowIndex] += countDelta;
durations[windowIndex] += durationDelta;
}
// 清空指定窗口的统计值
public void clear(int windowIndex) {
sessionCounts[windowIndex] = 0;
durations[windowIndex] = 0;
}
// 获取统计值
public long getSessionCount(int index) { return sessionCounts[index]; }
public long getDuration(int index) { return durations[index]; }
// 设置/获取最后事件时间
public void setLastEventTime(long time) { lastEventTime = time; }
public long getLastEventTime() { return lastEventTime; }
}
3.3 KeyedProcessFunction
public class UserStatsFunction extends KeyedProcessFunction<String, SessionEvent, UserStats> {
private transient ValueState<UserWindowState> state;
private static final int[] WINDOW_DURATIONS = {30 * 60 * 1000, 60 * 60 * 1000, 6 * 60 * 60 * 1000};
@Override
public void open(Configuration parameters) throws Exception {
ValueStateDescriptor<UserWindowState> descriptor =
new ValueStateDescriptor<>("userWindowState", UserWindowState.class);
state = getRuntimeContext().getState(descriptor);
}
@Override
public void processElement(SessionEvent event, Context ctx, Collector<UserStats> out) throws Exception {
UserWindowState currentState = state.value();
if (currentState == null) currentState = new UserWindowState();
// 更新所有窗口的统计值
for (int i = 0; i < 3; i++) {
currentState.update(i, 1, event.getSessionDuration());
}
currentState.setLastEventTime(event.getEventTime());
// 保存状态
state.update(currentState);
// 注册定时器到下一个整分钟
long nextTriggerTime = computeNextTriggerTime(event.getEventTime());
ctx.timerService().registerEventTimeTimer(nextTriggerTime);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<UserStats> out) throws Exception {
UserWindowState currentState = state.value();
if (currentState == null) return;
// 检查每个窗口是否过期
for (int i = 0; i < 3; i++) {
long windowDuration = WINDOW_DURATIONS[i];
long validStart = timestamp - windowDuration;
if (currentState.getLastEventTime() < validStart) {
currentState.clear(i);
}
}
// 保存更新后的状态
state.update(currentState);
// 输出统计结果
UserStats stats = new UserStats(
ctx.getCurrentKey(),
currentState.getSessionCount(0), currentState.getDuration(0),
currentState.getSessionCount(1), currentState.getDuration(1),
currentState.getSessionCount(2), currentState.getDuration(2)
);
out.collect(stats);
// 注册下个定时器
ctx.timerService().registerEventTimeTimer(timestamp + 60 * 1000);
}
// 计算下一个整分钟的触发时间
private long computeNextTriggerTime(long eventTime) {
long minutes = eventTime / 60000;
return (minutes + 1) * 60000;
}
}
3.4 Flink主程序
public class FlinkJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 输入数据流
DataStream<SessionEvent> input = env.addSource(new FlinkKafkaConsumer<>(...));
// 设置事件时间及Watermark
input = input.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor<SessionEvent>(Time.seconds(5)) {
@Override
public long extractTimestamp(SessionEvent element) {
return element.getEventTime();
}
}
);
// 处理逻辑
DataStream<UserStats> result = input
.keyBy(SessionEvent::getUserId)
.process(new UserStatsFunction());
// 输出结果
result.print();
env.execute("User Session Stats");
}
}
4.核心关键点说明
定时器机制：
每个用户每分钟触发一次定时器，检查窗口是否过期。
若用户最后一次事件时间超过窗口阈值，则清零统计值。
状态管理：
使用ValueState维护用户状态，包含三个窗口的会话数、时长和最后一次事件时间。
事件时间处理：
通过assignTimestampsAndWatermarks设置事件时间，确保窗口计算的准确性。
输出更新：
每分钟触发定时器时，输出当前用户的统计结果，即使统计值为零。
5.代码答疑
1.为什么需要两处注册？
首次触发的启动问题：
如果只有 onTimer 中的注册（即无 processElement 的注册），则定时器链可能无法启动。
假设用户长时间没有事件（如 2 小时未活跃），则 processElement 永远不会被调用，导致无法注册第一个定时器，定时器链无法启动。
processElement 的注册确保了当首次事件到达时，定时器链被启动。
持续触发的维持问题：
如果只有 processElement 的注册（即无 onTimer 的注册），则定时器链可能中断。
假设用户最后一次事件是在 14:23:45，则会注册 14:24:00 的定时器。
如果后续 没有新事件，则 processElement 不会被调用，无法注册 14:25:00 的定时器，导致定时器链在 14:24:00 后中断。
onTimer 的注册确保了即使没有新事件，定时器也能持续触发。
总结:
processElement 的注册是启动定时器链的必要条件（处理冷启动场景）。
onTimer 的注册是维持定时器链持续运行的核心逻辑（覆盖无新事件的情况）。
Flink 的内部机制会自动去重同一时间的定时器，因此 不会产生额外资源浪费。