Keyed State 详解：五种核心状态原语
说实话，刚接触 Flink 状态管理那会儿，我也被这五种状态搞晕过。ValueState、ListState、MapState……名字看着挺像，用起来却各有各的脾气。今天咱们就一个一个掰开揉碎了讲清楚。
先看一张总览图，把这五种状态的关系和定位理一理：
1. ValueState：最基础的单值状态
ValueState 存的就是一个值。一个 key 对应一个值，更新就覆盖，读取就返回当前值。说白了，它就是每个 key 的「私人储物柜」。
适用场景：记录最近一次事件时间、累计计数、状态开关等。
我在项目中遇到过这样一个需求：统计每个用户的最后登录时间。用 ValueState 就特别合适——每个用户 key 存一个 Long 类型的时间戳，来一条登录事件就 update 一次。
public class LastLoginTimeFunction extends RichFlatMapFunction<LoginEvent, String> {
private transient ValueState<Long> lastLoginState;
@Override
public void open(Configuration parameters) {
ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>("last-login", Types.LONG);
lastLoginState = getRuntimeContext().getState(descriptor);
}
@Override
public void flatMap(LoginEvent event, Collector<String> out) throws Exception {
Long lastTime = lastLoginState.value();
if (lastTime == null || event.getTimestamp() > lastTime) {
lastLoginState.update(event.getTimestamp());
out.collect("用户 " + event.getUserId() + " 最新登录时间更新");
}
}
}小技巧：ValueState 的 value() 方法返回 null 表示该 key 还没有状态。初始化时记得判空，不然容易 NPE。
2. ListState：有序列表状态
ListState 存的是一个列表。你可以往里面 add 元素，也可以 addAll 批量添加，还能 get 出整个 Iterable 遍历。每个 key 维护一个独立的列表。
嗯，这里要注意：ListState 内部是有序的，添加顺序就是存储顺序。但如果你需要去重，得自己控制。
适用场景：收集某个 key 的历史事件、缓存最近 N 条记录、批量处理前的数据缓冲。
我曾经做过一个「用户行为轨迹」的功能，需要把用户最近 100 条点击事件存下来。用 ListState 配合 TTL 就能优雅实现——超过 100 条时手动清理最早的，超过 TTL 的自动过期。
public class UserActionBuffer extends RichFlatMapFunction<ActionEvent, List<ActionEvent>> {
private transient ListState<ActionEvent> actionState;
private static final int MAX_SIZE = 100;
@Override
public void open(Configuration parameters) {
ListStateDescriptor<ActionEvent> descriptor =
new ListStateDescriptor<>("actions", Types.POJO(ActionEvent.class));
actionState = getRuntimeContext().getListState(descriptor);
}
@Override
public void flatMap(ActionEvent event, Collector<List<ActionEvent>> out) throws Exception {
actionState.add(event);
Iterable<ActionEvent> allActions = actionState.get();
List<ActionEvent> list = new ArrayList<>();
allActions.forEach(list::add);
// 超过上限时，移除最早的
while (list.size() > MAX_SIZE) {
list.remove(0);
}
out.collect(list);
}
}避坑指南：我曾经在生产环境遇到过 ListState 无限增长导致 OOM 的问题。原因是忘了设置 TTL，也没做大小限制。记住：ListState 不会自动裁剪，你必须自己控制列表长度。
3. MapState：键值对映射状态
MapState 就是每个 key 维护一个 Map。支持 put、get、remove、contains、entries、keys、values 等操作，跟 Java 的 HashMap 用起来差不多。
你想想看，如果需求是「统计每个用户每个商品的浏览次数」，用 MapState 就比用 ListState 优雅得多——商品 ID 做 key，浏览次数做 value，直接 get 和 put 就行。
适用场景：多维度的计数统计、配置映射、分组聚合的中间结果。
public class ProductViewCount extends RichFlatMapFunction<ViewEvent, Tuple2<String, Long>> {
private transient MapState<String, Long> viewCountState;
@Override
public void open(Configuration parameters) {
MapStateDescriptor<String, Long> descriptor =
new MapStateDescriptor<>("view-count", Types.STRING, Types.LONG);
viewCountState = getRuntimeContext().getMapState(descriptor);
}
@Override
public void flatMap(ViewEvent event, Collector<Tuple2<String, Long>> out) throws Exception {
String productId = event.getProductId();
Long count = viewCountState.get(productId);
if (count == null) {
count = 0L;
}
viewCountState.put(productId, ++count);
out.collect(new Tuple2<>(productId, count));
}
}性能提示：MapState 的 entries() 和 keys() 返回的是 Iterable，不是一次性加载到内存的。遍历大数据量时不用担心 OOM，但要注意迭代器使用完要 close。
4. ReducingState：自动归约状态
ReducingState 跟 ValueState 有点像，但它多了一个「归约函数」。每次 add 一个新值时，会自动用归约函数跟当前值合并。你不需要手动 update，add 完就完事了。
说白了，它就是帮你自动做增量聚合。比如累加、求最大值、求最小值，这些场景用 ReducingState 最省事。
适用场景：流式累加、滚动最大值/最小值、连续事件的归约统计。
public class MaxTemperature extends RichFlatMapFunction<SensorReading, Double> {
private transient ReducingState<Double> maxTempState;
@Override
public void open(Configuration parameters) {
ReducingStateDescriptor<Double> descriptor =
new ReducingStateDescriptor<>("max-temp",
(v1, v2) -> Math.max(v1, v2), // 归约函数：取较大值
Types.DOUBLE);
maxTempState = getRuntimeContext().getReducingState(descriptor);
}
@Override
public void flatMap(SensorReading reading, Collector<Double> out) throws Exception {
maxTempState.add(reading.getTemperature());
out.collect(maxTempState.get());
}
}注意：ReducingState 的输入类型和输出类型必须一致。因为归约函数的两个参数类型相同，返回值类型也相同。如果你需要输入输出类型不同，那就得用 AggregatingState。
5. AggregatingState：灵活的聚合状态
AggregatingState 是五种状态里最灵活的。它允许输入类型和输出类型不同，通过一个 AggregateFunction 来完成转换和聚合。
我个人的习惯是：只要输入输出类型不一致，或者聚合逻辑比较复杂，就用 AggregatingState。比如统计每个用户的平均消费金额——输入是单笔金额（Double），输出是平均值（Double），但中间需要维护总和和计数两个值。
适用场景：计算平均值、复杂指标统计、输入输出类型不同的聚合场景。
public class AveragePrice extends RichFlatMapFunction<TradeEvent, Double> {
private transient AggregatingState<Double, Double> avgPriceState;
@Override
public void open(Configuration parameters) {
AggregatingStateDescriptor<Double, Tuple2<Double, Integer>, Double> descriptor =
new AggregatingStateDescriptor<>("avg-price",
new AggregateFunction<Double, Tuple2<Double, Integer>, Double>() {
@Override
public Tuple2<Double, Integer> createAccumulator() {
return Tuple2.of(0.0, 0);
}
@Override
public Tuple2<Double, Integer> add(Double value, Tuple2<Double, Integer> acc) {
return Tuple2.of(acc.f0 + value, acc.f1 + 1);
}
@Override
public Double getResult(Tuple2<Double, Integer> acc) {
return acc.f1 == 0 ? 0.0 : acc.f0 / acc.f1;
}
@Override
public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a,
Tuple2<Double, Integer> b) {
return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
}
},
Types.TUPLE(Types.DOUBLE, Types.INT));
avgPriceState = getRuntimeContext().getAggregatingState(descriptor);
}
@Override
public void flatMap(TradeEvent event, Collector<Double> out) throws Exception {
avgPriceState.add(event.getPrice());
out.collect(avgPriceState.get());
}
}经验之谈：AggregatingState 的 merge 方法在状态合并时才会用到，比如 rescale 或 rebalance 之后。如果你的聚合逻辑不支持合并，记得在 merge 里抛异常或者做特殊处理。
五种状态对比一览
| 状态类型 | 数据结构 | 输入/输出类型 | 核心操作 | 典型场景 |
|---|---|---|---|---|
| ValueState | 单值 | 相同 | value() / update() | 记录最新值、开关状态 |
| ListState | 列表 | 相同 | add() / get() / update() | 事件缓冲、历史记录 |
| MapState | 键值对 | 相同 | put() / get() / remove() / entries() | 多维计数、配置映射 |
| ReducingState | 单值（自动归约） | 相同 | add() / get() | 累加、最大值、最小值 |
| AggregatingState | 单值（自定义聚合） | 可以不同 | add() / get() | 平均值、复杂指标 |
最后说一句：选哪种状态，核心就看两点——数据结构需求（单值/列表/映射）和聚合方式（手动/自动归约/自定义聚合）。搞清楚了这两点，你就能在项目中快速做出选择。