当前位置：首页 > java >正文

支付宝智能助理用户会话实时统计：Flink定时器与状态管理实战解析

java 2025/7/18 16:48:30

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};@Overridepublic void open(Configuration parameters) throws Exception {ValueStateDescriptor<UserWindowState> descriptor =new ValueStateDescriptor<>("userWindowState", UserWindowState.class);state = getRuntimeContext().getState(descriptor);}@Overridepublic 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);}@Overridepublic 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<>(...));// 设置事件时间及Watermarkinput = input.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SessionEvent>(Time.seconds(5)) {@Overridepublic long extractTimestamp(SessionEvent element) {return element.getEventTime();}});// 处理逻辑DataStream<UserStats> result = input.keyBy(SessionEvent::getUserId).process(new UserStatsFunction());// 输出结果result.print();env.execute("User Session Stats");}
}