3. 状态数据结构

状态编程的核心，说白了就是选对数据结构。

Flink 提供了五种内置的状态类型。我刚开始接触时也觉得种类有点多，但用久了你会发现——每种状态都有它不可替代的场景。选对了，代码简洁、性能好；选错了，要么逻辑绕弯子，要么内存爆炸。

这一节，我们逐个拆解。

ValueState：最基础的状态容器

ValueState 存的是单个值。你可以把它理解成一个 key 级别的变量。

它适合什么场景？比如统计每个用户的当前登录状态、记录最近一次操作时间、保存一个计数器。说白了，就是「一个 key 对应一个值」的场景。

// 定义一个 ValueState，记录用户最近一次登录时间戳
private transient ValueState<Long> lastLoginTime;

// 在 open 方法中初始化
lastLoginTime = getRuntimeContext()
    .getState(new ValueStateDescriptor<>("last-login", Types.LONG));

// 使用
Long previous = lastLoginTime.value();
if (previous == null) {
    // 首次登录
    lastLoginTime.update(System.currentTimeMillis());
} else {
    // 判断是否超过 24 小时
    if (System.currentTimeMillis() - previous > 86400000L) {
        // 触发重新登录逻辑
    }
}

我在项目中遇到过一个问题：ValueState 的默认值处理。很多人以为 value() 返回 0 或空字符串，其实它返回 null。如果你不做判空，直接拿去做算术运算，NPE 就来了。

ListState：有序集合，适合追加场景

ListState 维护一个元素列表。每个 key 对应一个 List。

它最擅长处理「不断追加」的数据。比如记录用户的所有操作日志、保存一批待处理的事件 ID、缓存最近 N 条交易记录。

// 定义 ListState，保存用户最近 10 条搜索记录
private transient ListState<String> recentSearches;

recentSearches = getRuntimeContext()
    .getListState(new ListStateDescriptor<>("recent-searches", Types.STRING));

// 追加新记录
recentSearches.add(searchKeyword);

// 获取所有记录
Iterable<String> all = recentSearches.get();
// 注意：get() 返回的是 Iterable，不是 List
// 如果需要 List，需要自己遍历转换

嗯，这里要注意一个坑：ListState 的 get() 返回的是 Iterable，不是 java.util.List。你不能直接调用 get(0) 或者 size()。我早期写代码时踩过这个坑，后来养成了习惯——需要随机访问时，先转成 ArrayList。

MapState：KV 结构，适合索引场景

MapState 维护一个 key-value 映射。每个 Flink key 对应一个 Map。

它适合什么？用户配置项（key 是配置名，value 是配置值）、商品库存（key 是 SKU，value 是库存量）、会话中的属性集合。

// 定义 MapState，保存用户的购物车信息
// key: 商品ID, value: 数量
private transient MapState<String, Integer> cartItems;

cartItems = getRuntimeContext()
    .getMapState(new MapStateDescriptor<>("cart", Types.STRING, Types.INT));

// 添加商品
cartItems.put("item_001", 2);

// 获取某个商品数量
Integer count = cartItems.get("item_001");

// 遍历所有商品
for (Map.Entry<String, Integer> entry : cartItems.entries()) {
    // 处理每个商品
}

// 判断是否存在
boolean exists = cartItems.contains("item_002");

MapState 的 API 设计得很贴心。contains()、keys()、values()、entries() 都有，用起来跟标准 Map 差不多。但有一点不同——它没有 size() 方法。想统计元素个数？得自己遍历计数。

ReducingState：自动聚合，无需手动合并

ReducingState 是 ValueState 的升级版。你给它一个 ReduceFunction，每次 add 时自动做聚合。

它适合「只关心聚合结果，不关心中间数据」的场景。比如统计每个用户的累计消费金额、计算当前最高温度、合并事件流中的重复数据。

// 定义 ReducingState，统计用户累计消费金额
private transient ReducingState<Double> totalSpent;

totalSpent = getRuntimeContext()
    .getReducingState(new ReducingStateDescriptor<>(
        "total-spent",
        (v1, v2) -> v1 + v2,  // 累加逻辑
        Types.DOUBLE
    ));

// 每次消费时调用
totalSpent.add(99.9);
totalSpent.add(50.0);

// 获取累计值
Double currentTotal = totalSpent.get();  // 返回 149.9

你想想看，如果用 ValueState 实现同样的功能，你得手动写：oldValue = state.value(); newValue = oldValue + newAmount; state.update(newValue)。ReducingState 帮你省掉了这三行样板代码。

AggregatingState：更灵活的聚合

AggregatingState 是 ReducingState 的进一步泛化。它允许你维护一个中间累加器类型，最终输出一个不同的结果类型。

为什么需要这个？举个例子：你要计算每个用户的平均消费金额。中间状态需要维护「总和」和「次数」两个值，但最终输出只需要「平均值」一个值。AggregatingState 正好解决这个问题。

// 定义 AggregatingState，计算用户平均消费
private transient AggregatingState<Double, Double> avgSpent;

avgSpent = getRuntimeContext()
    .getAggregatingState(new AggregatingStateDescriptor<>(
        "avg-spent",
        new AggregateFunction<Double, Tuple2<Double, Long>, Double>() {
            @Override
            public Tuple2<Double, Long> createAccumulator() {
                return Tuple2.of(0.0, 0L);  // (总和, 次数)
            }

            @Override
            public Tuple2<Double, Long> add(Double value, Tuple2<Double, Long> acc) {
                return Tuple2.of(acc.f0 + value, acc.f1 + 1);
            }

            @Override
            public Double getResult(Tuple2<Double, Long> acc) {
                return acc.f1 == 0 ? 0.0 : acc.f0 / acc.f1;
            }

            @Override
            public Tuple2<Double, Long> merge(Tuple2<Double, Long> a, 
                                               Tuple2<Double, Long> b) {
                return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
            }
        },
        Types.TUPLE(Types.DOUBLE, Types.LONG)
    ));

// 使用
avgSpent.add(100.0);
avgSpent.add(200.0);
Double avg = avgSpent.get();  // 返回 150.0

说实话，AggregatingState 的代码量看起来比 ReducingState 多。但它的灵活性是其他状态类型无法替代的。我在做实时用户画像时，经常用它来维护各种复合指标。

五种状态类型对比

状态类型	数据结构	核心操作	典型场景	注意事项
ValueState	单个值	value(), update()	计数器、状态标记	默认值为 null
ListState	元素列表	add(), get(), clear()	操作日志、事件缓存	注意数据量膨胀
MapState	KV 映射	put(), get(), contains()	配置项、购物车	无 size() 方法
ReducingState	聚合值	add(), get()	累计求和、最大值	需满足结合律
AggregatingState	中间累加器	add(), get()	平均值、复合指标	代码量较大

如何选择？我的经验法则

选状态类型其实有个简单的判断流程：

1. 只存一个值？ → ValueState
2. 需要存多个值，且不断追加？ → ListState
3. 需要按 key 索引？ → MapState
4. 只需要聚合结果，不需要中间数据？ → ReducingState
5. 聚合逻辑复杂，中间类型和输出类型不同？ → AggregatingState

我个人习惯是：能用 ReducingState 就不用 ValueState 手动聚合，能用 AggregatingState 就不用自己维护累加器。Flink 内置的优化比你手写的循环要高效得多。

状态序列化与性能考量

选对了状态类型只是第一步。序列化方式直接影响性能。

Flink 默认使用 Java 序列化，但性能很差。我建议你：

• 优先使用 Flink 内置的 TypeSerializer，比如 Types.STRING、Types.LONG、Types.POJO
• 避免使用 Kryo 序列化，除非万不得已。Kryo 虽然方便，但序列化后的数据体积大，GC 压力也大
• 自定义 POJO 时，实现 Serializable 接口，并确保有无参构造器

状态 TTL：别让状态永远活着

所有状态类型都支持 TTL（Time-To-Live）配置。这个功能太重要了。

没有 TTL 的状态，就像没有垃圾回收的内存——迟早会爆。我见过太多线上事故，都是因为状态只增不减。

// 配置 ValueState 的 TTL
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))           // 存活 7 天
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)  // 写入时更新过期时间
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)  // 不返回过期数据
    .build();

ValueStateDescriptor<Long> descriptor = 
    new ValueStateDescriptor<>("state-with-ttl", Types.LONG);
descriptor.enableTimeToLive(ttlConfig);

配置 TTL 时有几个要点：

• UpdateType：OnCreateAndWrite 是常用选项，每次写入都刷新过期时间
• StateVisibility：NeverReturnExpired 更安全，不会读到过期数据
• TTL 不是精确的：过期数据会在下次访问或 checkpoint 时被清理，不是到点就删