3. Keyed State 入门：五种状态原语的概念与使用场景

好，咱们进入 Keyed State 的世界。说实话，Flink 的状态管理里，Keyed State 是出场率最高的，也是你每天写作业逻辑时打交道最多的东西。我刚开始学 Flink 时，总觉得状态这玩意儿很玄乎，后来踩了几个坑才明白——它其实就是跟 key 绑定的“私有变量”。

Keyed State 有五种内置原语：ValueState、ListState、MapState、ReducingState、AggregatingState。每种都有自己的脾气和适用场景。下面我一个一个说，顺便聊聊我在项目中是怎么选型的。

核心认知：Keyed State 必须用在 KeyedStream 上。也就是说，你得先 .keyBy()，才能使用这些状态。否则运行时直接报错，别问我怎么知道的……

3.1 ValueState：最简单的“单值”状态

ValueState 存的就是一个值。每个 key 对应一个值，更新就是覆盖。说白了，它就像 Map<K, V> 里的那个 V。

使用场景：

  • 记录每个用户的最后登录时间
  • 统计每个商品的累计点击量（只保留最新值）
  • 保存每个传感器的最近一次温度读数

我记得有一次做实时风控，需要记录每个用户最近一笔交易的金额。用 ValueState 就特别合适——每次来新交易，直接 update 覆盖旧值，查询时用 value() 方法取出来就行。

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

// 在 open() 中初始化
@Override
public void open(Configuration parameters) {
    ValueStateDescriptor<Long> descriptor = 
        new ValueStateDescriptor<>(
            "lastLoginTime",   // 状态名称
            Types.LONG         // 类型信息
        );
    lastLoginTime = getRuntimeContext().getState(descriptor);
}

// 在 processElement() 中使用
@Override
public void processElement(
        LoginEvent event, 
        Context ctx, 
        Collector<String> out) throws Exception {
    
    Long previousTime = lastLoginTime.value();  // 读取
    if (previousTime != null) {
        // 做业务判断...
    }
    lastLoginTime.update(event.getTimestamp());  // 更新
}

个人习惯：我一般把 ValueStateDescriptor 的 name 定义成常量，方便后续维护。另外，类型信息尽量用 Types 工具类指定，别偷懒用 TypeInformation.of() 去推断，容易出序列化问题。

3.2 ListState：有序的“列表”状态

ListState 存的是一个列表。每个 key 对应一个 List，你可以往里面 add 元素，也可以 addAll 批量添加。取出来是一个 Iterable，遍历着用。

使用场景：

  • 收集某个用户最近 10 条浏览记录
  • 缓存一批数据，等攒够一定数量再批量处理
  • 记录某个订单的所有状态变更历史

我曾经做过一个“用户行为序列”分析的任务。需要把用户连续 5 次点击事件攒在一起，判断是否有异常模式。用 ListState 就很方便——每次来事件就 add，然后判断 size 是否达到 5，到了就取出来处理，再 clear 清空。

private transient ListState<String> clickHistory;

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

@Override
public void processElement(
        ClickEvent event, 
        Context ctx, 
        Collector<String> out) throws Exception {
    
    clickHistory.add(event.getPageId());
    
    Iterable<String> allClicks = clickHistory.get();
    // 遍历处理...
    
    // 如果列表太长，可以清理
    // clickHistory.clear();
}

注意：ListState 没有“按索引删除”的方法。如果你想删除某个特定元素，得把整个列表取出来，在内存里操作完再重新 update。数据量大时小心 OOM。

3.3 MapState：键值对“字典”状态

MapState 存的是一个 Map。每个 key 对应一个 Map<UK, UV>，支持 put、get、remove、contains 等操作。说白了，它就是嵌套的键值对。

使用场景：

  • 维护每个用户的购物车（商品ID -> 数量）
  • 记录每个设备的配置参数（参数名 -> 参数值）
  • 缓存每个渠道的实时费率表（费率类型 -> 费率值）

我印象很深的一个项目是实时推荐系统。需要为每个用户维护一个“商品偏好分数”的 Map，key 是商品类目，value 是偏好分值。用 MapState 就特别自然——更新某个类目的分数时直接 put，查询时 get，判断某个类目是否存在用 contains。

private transient MapState<String, Double> preferenceScores;

@Override
public void open(Configuration parameters) {
    MapStateDescriptor<String, Double> descriptor = 
        new MapStateDescriptor<>(
            "preferenceScores",
            Types.STRING,   // Map 的 key 类型
            Types.DOUBLE    // Map 的 value 类型
        );
    preferenceScores = getRuntimeContext().getMapState(descriptor);
}

@Override
public void processElement(
        UserAction action, 
        Context ctx, 
        Collector<String> out) throws Exception {
    
    // 更新偏好分数
    Double current = preferenceScores.get(action.getCategory());
    if (current == null) {
        current = 0.0;
    }
    preferenceScores.put(action.getCategory(), current + action.getScore());
    
    // 遍历所有偏好
    for (Map.Entry<String, Double> entry : preferenceScores.entries()) {
        // 处理...
    }
}

避坑指南：MapState 的 entries()、keys()、values() 返回的都是 Iterable，不是 Collection。你不能直接调用 size() 方法——得自己遍历计数。另外，MapState 的 key 和 value 的类型必须可序列化。

3.4 ReducingState：自动“聚合”的状态

ReducingState 跟 ValueState 有点像，但它多了一个“聚合函数”。每次你 add 一个新值，它就会自动跟当前值做聚合，结果存回去。你不需要手动 update。

使用场景：

  • 实时累加求和（比如每个用户的累计消费金额）
  • 实时求最大值/最小值（比如每个传感器的最高温度）
  • 字符串拼接（比如收集所有异常码）

说白了，ReducingState 就是帮你省掉了“读旧值 -> 计算 -> 写新值”这三步，你只管 add 就行。我一般在做简单累加时特别喜欢用它，代码干净很多。

private transient ReducingState<Long> totalAmount;

@Override
public void open(Configuration parameters) {
    ReducingStateDescriptor<Long> descriptor = 
        new ReducingStateDescriptor<>(
            "totalAmount",
            new ReduceFunction<Long>() {
                @Override
                public Long reduce(Long value1, Long value2) {
                    return value1 + value2;  // 累加逻辑
                }
            },
            Types.LONG
        );
    totalAmount = getRuntimeContext().getReducingState(descriptor);
}

@Override
public void processElement(
        OrderEvent event, 
        Context ctx, 
        Collector<String> out) throws Exception {
    
    totalAmount.add(event.getAmount());  // 自动累加
    Long currentTotal = totalAmount.get();  // 读取当前累加值
}

关键区别：ReducingState 的输入类型和输出类型必须相同。因为 reduce 函数的输入是两个同类型的值，输出也是同类型。如果你需要输入输出类型不同，那就得用 AggregatingState。

3.5 AggregatingState：更灵活的“聚合”状态

AggregatingState 是 ReducingState 的升级版。它的输入类型（IN）、中间累加类型（ACC）、输出类型（OUT）可以各不相同。你需要实现 AggregateFunction 接口，定义 add、merge、getResult 等方法。

使用场景：

  • 计算平均值（输入是数值，输出是字符串或 Double）
  • 统计事件序列（输入是事件对象，输出是统计报告）
  • 复杂的数据转换和聚合

我记得有个需求：统计每个用户每分钟的“事件频率”。输入是事件时间戳，中间累加器是 count 和 sum，输出是格式化的字符串 "频率: X 次/分钟"。用 AggregatingState 就特别优雅。

private transient AggregatingState<Long, String> frequencyState;

@Override
public void open(Configuration parameters) {
    AggregatingStateDescriptor<Long, AverageAccum, String> descriptor = 
        new AggregatingStateDescriptor<>(
            "frequencyState",
            new AggregateFunction<Long, AverageAccum, String>() {
                @Override
                public AverageAccum createAccumulator() {
                    return new AverageAccum();
                }

                @Override
                public AverageAccum add(Long value, AverageAccum acc) {
                    acc.sum += value;
                    acc.count++;
                    return acc;
                }

                @Override
                public String getResult(AverageAccum acc) {
                    return "频率: " + (acc.count > 0 ? acc.sum / acc.count : 0) + " 次/分钟";
                }

                @Override
                public AverageAccum merge(AverageAccum a, AverageAccum b) {
                    a.count += b.count;
                    a.sum += b.sum;
                    return a;
                }
            },
            Types.POJO(AverageAccum.class)
        );
    frequencyState = getRuntimeContext().getAggregatingState(descriptor);
}

// 辅助类
public static class AverageAccum {
    public long sum = 0;
    public long count = 0;
}

注意：AggregatingState 的中间累加器（ACC）必须是一个可序列化的 POJO 或基本类型。我见过有人用不可序列化的对象做累加器，结果运行时直接炸了。另外，merge 方法在状态合并时才会用到，比如 rescale 或恢复时。

3.6 五种状态原语对比

为了方便你选型，我整理了一张对比表。你写代码前可以瞄一眼，省得选错类型回头改代码。

状态类型 数据结构 输入/输出类型 核心操作 典型场景
ValueState 单值 相同 value(), update(), clear() 记录最新值、计数器
ListState 列表 相同 add(), addAll(), get(), clear() 缓存一批数据、历史记录
MapState 键值对 Key/Value 可不同 put(), get(), remove(), contains(), entries() 字典、配置、购物车
ReducingState 单值（自动聚合） 相同 add(), get(), clear() 累加、求最大/最小
AggregatingState 单值（灵活聚合） 可不同 add(), get(), clear() 平均值、复杂统计

3.7 本章知识体系

下面这张图帮你梳理了五种 Keyed State 的关系和选型思路。你可以保存下来，写代码时当参考。

Keyed State 五种原语选型指南 Keyed State ValueState ListState MapState ReducingState AggregatingState 选型口诀： • 只存一个值 → ValueState • 需要列表/队列 → ListState • 需要嵌套键值对 → MapState • 简单聚合（同类型） → ReducingState • 复杂聚合（不同类型） → AggregatingState 💡 所有 Keyed State 都必须在 keyBy() 之后使用。 💡 状态名称（descriptor 的 name）在同一算子内必须唯一。 Keyed State 选型决策树

3.8 如何选择？我的经验法则

说实话，刚开始接触这五种状态时，我也经常纠结选哪个。后来总结了一套简单的判断逻辑：

  1. 先看数据结构：单值？列表？Map？这个决定了 80% 的选择。
  2. 再看聚合需求：如果只是存和取，ValueState 就够了。如果需要自动聚合，再看输入输出类型是否相同。
  3. 最后看性能：ListState 和 MapState 在数据量大时要注意内存。ReducingState 和 AggregatingState 因为只保留聚合结果，内存占用更小。
← 上一章 📖 返回目录 下一章 →