2. 状态类型详解（上）：Keyed State 之 ValueState、ListState、MapState 原理与使用场景

好，咱们进入状态类型的正题。Flink 的状态分两大类：Keyed State 和 Operator State。这一章我们先啃 Keyed State 里最常用的三个：ValueState、ListState、MapState。

说白了，Keyed State 就是「每个 key 一份状态」。你想想看，在分布式系统里，数据按 key 分区后，每个并行子任务只处理一部分 key。那每个 key 自己的状态，自然得独立管理。Flink 帮我们封装好了，你只管用。

2.1 ValueState：最简单的状态，最常用的场景

ValueState 就是存一个值。可以是任何类型——Long、String、Pojo，甚至你自定义的对象。每次只存一个，更新就覆盖。

使用场景：

• 计数器：统计每个用户的点击次数
• 最后一条记录：记录每个传感器的最新读数
• 聚合中间结果：累加器、最大值、最小值

我在项目中遇到过这样一个需求：统计每个广告位的曝光次数，然后每满 100 次触发一次回调。用 ValueState 存一个 Long 计数器，每次来事件就 update，到了阈值就清空。代码非常干净。

// 定义 ValueState
private transient ValueState<Long> countState;

// 在 open() 中初始化
@Override
public void open(Configuration parameters) {
    ValueStateDescriptor<Long> descriptor = 
        new ValueStateDescriptor<>("clickCount", Types.LONG);
    countState = getRuntimeContext().getState(descriptor);
}

// 处理逻辑
@Override
public void processElement(String value, Context ctx, Collector<String> out) {
    Long currentCount = countState.value();  // 读取
    if (currentCount == null) {
        currentCount = 0L;
    }
    currentCount += 1;
    countState.update(currentCount);         // 更新
    if (currentCount >= 100) {
        out.collect("触发回调：" + value);
        countState.clear();                  // 清空
    }
}

2.2 ListState：有序列表，适合存「历史」

ListState 存的是一个元素列表。每个 key 对应一个 List，你可以往里面追加元素，也可以遍历所有元素。

使用场景：

• 最近 N 条记录：比如每个用户最近 10 次登录 IP
• 事件序列：检测模式匹配时，缓存一段时间内的事件
• 批量输出：攒够一批数据再发送

嗯，这里要注意：ListState 不是 java.util.List，它有自己的 API。你不能直接 addAll 一个集合，得用 Iterable 遍历。

// 定义 ListState
private transient ListState<String> loginHistoryState;

@Override
public void open(Configuration parameters) {
    ListStateDescriptor<String> descriptor = 
        new ListStateDescriptor<>("loginHistory", Types.STRING);
    loginHistoryState = getRuntimeContext().getListState(descriptor);
}

@Override
public void processElement(String userId, Context ctx, Collector<String> out) {
    // 追加当前登录 IP
    loginHistoryState.add("192.168.1." + System.currentTimeMillis() % 255);
    
    // 遍历所有历史记录
    Iterable<String> history = loginHistoryState.get();
    StringBuilder sb = new StringBuilder("登录历史: ");
    for (String ip : history) {
        sb.append(ip).append(" -> ");
    }
    out.collect(sb.toString());
}

2.3 MapState：键值映射，灵活的数据字典

MapState 就是一个 key-value 映射表。每个 key 对应一个 Map，你可以按子 key 存取。说白了，就是「状态的二维结构」。

使用场景：

• 配置表：每个用户的自定义配置项
• 去重：用 MapState 存已经处理过的 ID，value 可以存时间戳
• 多维度聚合：按不同维度分别统计

MapState 的 API 很丰富，支持 containsKey、get、put、remove、entries、keys、values 等操作。比 ListState 灵活得多。

// 定义 MapState
private transient MapState<String, Double> configState;

@Override
public void open(Configuration parameters) {
    MapStateDescriptor<String, Double> descriptor = 
        new MapStateDescriptor<>("userConfig", Types.STRING, Types.DOUBLE);
    configState = getRuntimeContext().getMapState(descriptor);
}

@Override
public void processElement(String userId, Context ctx, Collector<String> out) {
    // 检查是否有配置
    if (configState.contains("threshold")) {
        Double threshold = configState.get("threshold");
        // 使用配置值
    }
    // 更新配置
    configState.put("threshold", 0.8);
    // 遍历所有配置
    for (Map.Entry<String, Double> entry : configState.entries()) {
        // 处理每个配置项
    }
}

2.4 三种状态的对比与选型

特性	ValueState	ListState	MapState
数据结构	单个值	有序列表	键值映射
读写方式	value() / update()	add() / get() / update()	put() / get() / contains()
遍历支持	不支持	Iterable	keys() / values() / entries()
典型场景	计数器、最后一条记录	历史记录、事件序列	配置表、去重、多维度聚合
状态大小	小（单值）	中（列表）	大（映射表）
序列化开销	低	中	高（取决于条目数）

选型其实不难：

• 只需要存一个值？用 ValueState。
• 需要按顺序存多个值？用 ListState。
• 需要按子 key 索引？用 MapState。

我个人习惯是：能用 ValueState 解决的，绝不用 ListState。因为 ValueState 的序列化开销最小，性能最好。MapState 虽然灵活，但状态后端对它的优化不如 ValueState 成熟，尤其是 RocksDB 后端下，MapState 的每次 put/get 都可能涉及序列化。

2.5 状态 TTL：别让状态「永远活着」

不管用哪种状态，都要考虑 TTL（Time-To-Live）。默认情况下，状态是永久存在的。你想想看，如果 key 不再来了，状态还占着内存或磁盘，这不是浪费吗？

// 配置 TTL
ValueStateDescriptor<Long> descriptor = 
    new ValueStateDescriptor<>("clickCount", Types.LONG);

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(1))           // 存活 1 天
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)  // 写入时更新 TTL
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)  // 过期不返回
    .build();

descriptor.enableTimeToLive(ttlConfig);

TTL 的配置项不少，我常用的就三个：

• 存活时间：多久没更新就过期
• 更新策略：OnCreateAndWrite（写入时续期）还是 OnReadAndWrite（读取也续期）
• 可见性：过期后是否还能读到（一般选 NeverReturnExpired）

好了，ValueState、ListState、MapState 的原理和用法就这些。下一类状态——ReducingState 和 AggregatingState，以及 Operator State，我们后面再聊。记住一句话：选对状态类型，比优化代码更重要。