Keyed State 详解：五种状态原语的使用与原理

说实话，刚接触 Flink 状态管理那会儿，我也被这五种状态原语搞得有点晕。ValueState、ListState、MapState、ReducingState、AggregatingState——名字看着都挺像，用起来却各有各的脾气。今天我就把这五个家伙掰开揉碎，结合我踩过的坑，给你讲明白。

5.1 ValueState：最基础的状态容器

ValueState 说白了就是一个键值对，存一个值。每个 key 对应一个独立的值。我最早用它来记录用户的最后登录时间，简单粗暴。

// 定义一个 ValueState，记录用户累计消费金额
private transient ValueState<Double> totalSpendState;

@Override
public void open(Configuration parameters) {
    ValueStateDescriptor<Double> descriptor = 
        new ValueStateDescriptor<>(
            "total-spend",          // 状态名称
            TypeInformation.of(new TypeHint<Double>() {})  // 类型信息
        );
    totalSpendState = getRuntimeContext().getState(descriptor);
}

@Override
public void processElement(String key, Context ctx, Collector<String> out) {
    Double current = totalSpendState.value();
    if (current == null) {
        current = 0.0;
    }
    totalSpendState.update(current + event.getAmount());
}

5.2 ListState：有序集合的妙用

ListState 维护一个有序列表。每个 key 可以存多个元素，支持追加、遍历、清空。我常用它来缓存用户最近 100 条操作记录，做行为分析。

// 缓存用户最近的操作记录
private transient ListState<String> recentActionsState;

@Override
public void open(Configuration parameters) {
    ListStateDescriptor<String> descriptor = 
        new ListStateDescriptor<>(
            "recent-actions",
            TypeInformation.of(new TypeHint<String>() {})
        );
    recentActionsState = getRuntimeContext().getListState(descriptor);
}

// 只保留最近 100 条
public void addAction(String action) throws Exception {
    recentActionsState.add(action);
    Iterable<String> all = recentActionsState.get();
    List<String> list = new ArrayList<>();
    all.forEach(list::add);
    if (list.size() > 100) {
        recentActionsState.clear();
        list.subList(list.size() - 100, list.size())
            .forEach(s -> {
                try {
                    recentActionsState.add(s);
                } catch (Exception e) {
                    // 异常处理
                }
            });
    }
}

5.3 MapState：键值对集合的王者

MapState 就是分布式版的 HashMap。每个 key 对应一个 Map，支持 put、get、contains、entries 等操作。我在做实时推荐系统时，用它来存储用户的商品浏览记录，key 是商品 ID，value 是浏览次数。

// 记录用户对每个商品的浏览次数
private transient MapState<String, Long> productViewCountState;

@Override
public void open(Configuration parameters) {
    MapStateDescriptor<String, Long> descriptor = 
        new MapStateDescriptor<>(
            "product-view-count",
            TypeInformation.of(new TypeHint<String>() {}),
            TypeInformation.of(new TypeHint<Long>() {})
        );
    productViewCountState = getRuntimeContext().getMapState(descriptor);
}

public void recordView(String productId) throws Exception {
    Long count = productViewCountState.get(productId);
    if (count == null) {
        count = 0L;
    }
    productViewCountState.put(productId, count + 1);
}

操作	方法	说明
获取值	get(key)	返回指定 key 的值，不存在返回 null
添加/更新	put(key, value)	覆盖已有值
批量添加	putAll(Map)	一次性添加多个键值对
删除	remove(key)	删除指定 key
遍历	entries() / keys() / values()	返回 Iterable
判断存在	contains(key)	返回 boolean

5.4 ReducingState：增量聚合的利器

ReducingState 和 ValueState 有点像，但它自带一个 ReduceFunction。每次 add 一个新值，都会和当前值做聚合。我习惯用它来做实时计数，比如统计每个商品的实时销量。

// 实时销量统计
private transient ReducingState<Long> salesCountState;

@Override
public void open(Configuration parameters) {
    ReducingStateDescriptor<Long> descriptor = 
        new ReducingStateDescriptor<>(
            "sales-count",
            (v1, v2) -> v1 + v2,  // 累加逻辑
            TypeInformation.of(new TypeHint<Long>() {})
        );
    salesCountState = getRuntimeContext().getReducingState(descriptor);
}

public void recordSale(Long amount) throws Exception {
    salesCountState.add(amount);
    Long total = salesCountState.get();  // 直接拿到聚合结果
}

5.5 AggregatingState：最灵活的聚合器

AggregatingState 是 ReducingState 的升级版。它允许你定义输入类型、中间累加器类型和输出类型。说白了，你可以把原始数据先转换，再聚合，最后输出你想要的结果。

// 计算每个用户的平均消费金额
private transient AggregatingState<Double, Double> avgSpendState;

@Override
public void open(Configuration parameters) {
    AggregatingStateDescriptor<Double, AvgAccumulator, Double> descriptor = 
        new AggregatingStateDescriptor<>(
            "avg-spend",
            new AggregateFunction<Double, AvgAccumulator, Double>() {
                @Override
                public AvgAccumulator createAccumulator() {
                    return new AvgAccumulator();
                }

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

                @Override
                public Double getResult(AvgAccumulator acc) {
                    return acc.count == 0 ? 0.0 : acc.sum / acc.count;
                }

                @Override
                public AvgAccumulator merge(AvgAccumulator a, AvgAccumulator b) {
                    a.sum += b.sum;
                    a.count += b.count;
                    return a;
                }
            },
            TypeInformation.of(new TypeHint<AvgAccumulator>() {})
        );
    avgSpendState = getRuntimeContext().getAggregatingState(descriptor);
}

// 自定义累加器
public static class AvgAccumulator {
    double sum = 0.0;
    long count = 0;
}

5.6 五种状态原语对比与选型

为了让你看得更清楚，我画了一张对比图。这张图展示了每种状态原语的核心特征和适用场景。

5.7 状态后端对状态原语的影响

嗯，这里要注意。不同的状态后端，对这五种状态原语的支持和性能表现是不一样的。我遇到过有人用 RocksDB 后端，结果 MapState 的 entries() 遍历慢得离谱。

状态后端	ValueState	ListState	MapState	ReducingState	AggregatingState
HashMapStateBackend	极快（内存操作）	快	快	极快	快
RocksDBStateBackend	快（序列化/反序列化）	中等（追加操作需序列化整个列表）	中等（随机访问快，遍历慢）	快	中等

5.8 状态 TTL：别让状态无限膨胀

状态不设 TTL，就像家里不扔垃圾，迟早会堆满。Flink 1.8 之后支持状态 TTL，可以自动清理过期数据。我个人习惯，所有状态都设 TTL，哪怕设个很大的值，也比不设强。

// 给 ValueState 设置 TTL，过期后自动清理
ValueStateDescriptor<Double> descriptor = 
    new ValueStateDescriptor<>("total-spend", Double.class);

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

descriptor.enableTimeToLive(ttlConfig);