乔人尹 - State
type: channel
link: https://underestimated.cn/tag/State/
items: Flink从入门到放弃③—State与容错：有状态计算的基石

Flink从入门到放弃③—State与容错：有状态计算的基石
type: item
date: 2023-08-12T03:36:00+00:00
description: 一、前言：为什么需要 State？1.1 无状态 vs 有状态graph LR subgraph "无状态计算" A1[输入1] --> B1[处理] --> C1[输出1] A2[输入2] --> B2[处理] --> C2[输出2] A3[输入3] --> B3[处理] --> C3[输出3] endgraph LR subgraph "有状态计算" A1[输入1] --> B1[处理 + 状态更新] B1 --> STATE[(State)] STATE --> B2[处理 + 状态更新] A2[输入2] --> B2 B2 --> STATE STATE --> B3[处理 + 状态更新] A3[输入3] --> B3 end1.2 哪些场景需要 State？graph TB A[需要State的场景] --> B[聚合计算] A --> C[去重] A --> D[窗口计算] A --> E[机器学习模型] A --> F[CEP模式匹配] B --> B1["SUM/COUNT/AVG"] C --> C1["UV统计"] D --> D1["窗口内数据"] E --> E1["模型参数"] F --> F1["历史事件序列"]1.3 Flink State 的核心价值graph LR A[Flink State] --> B[高效访问] A --> C[容错保证] A --> D[精确一次] B --> B1["本地内存/RocksDB毫秒级访问"] C --> C1["Checkpoint机制故障自动恢复"] D --> D1["Exactly-Once状态一致性"]二、State 类型详解2.1 State 分类graph TB subgraph "Flink State 分类" A[Managed State托管状态] --> A1[Keyed State] A --> A2[Operator State] B[Raw State原始状态] --> B1["用户自行管理不推荐使用"] end A1 --> K1[ValueState] A1 --> K2[ListState] A1 --> K3[MapState] A1 --> K4[ReducingState] A1 --> K5[AggregatingState] A2 --> O1[ListState] A2 --> O2[BroadcastState]2.2 Keyed State 详解graph TB subgraph "Keyed State 特点" A["按 Key 分区"] B["每个 Key 独立状态"] C["只能在 KeyedStream 后使用"] endKeyed State 类型：类型说明适用场景ValueState单个值最新值、计数器ListState值列表历史记录MapState键值对多维度状态ReducingState聚合值（同类型）累加器AggregatingState聚合值（不同类型）复杂聚合代码示例：public class StatefulFunction extends KeyedProcessFunction<String, Event, Result> { // 1. ValueState - 单个值 private ValueState<Long> countState; // 2. ListState - 列表 private ListState<Event> historyState; // 3. MapState - Map private MapState<String, Long> categoryCountState; // 4. ReducingState - 归约 private ReducingState<Long> sumState; @Override public void open(Configuration parameters) throws Exception { // 初始化 ValueState ValueStateDescriptor<Long> countDesc = new ValueStateDescriptor<>("count", Long.class); countState = getRuntimeContext().getState(countDesc); // 初始化 ListState ListStateDescriptor<Event> historyDesc = new ListStateDescriptor<>("history", Event.class); historyState = getRuntimeContext().getList…