1. Flink 状态管理概述：什么是有状态计算、状态的分类与存储后端

各位同学，咱们开始聊 Flink 状态管理。说实话，我刚接触流计算那会儿，对「状态」这个概念也迷糊了好一阵子。你想想看，批处理算完就完事了，数据来了算一遍，结果落盘，下一批再来。但流处理不一样——数据一条一条地来，你得记住之前算到哪儿了，不然怎么算窗口？怎么算累计值？

嗯，这就是有状态计算的核心。

1.1 什么是有状态计算

有状态计算，说白了就是算子能记住历史数据。比如你算一个小时的 PV，来了第一条数据是 1，第二条来了变成 2，这个「当前累计值」就是状态。

我举个例子。你在做电商的实时大屏，要统计每分钟的订单金额。如果没有状态，每来一条订单你都得从头扫一遍历史数据——这不现实。有了状态，你只需要维护一个累加器，来一条加一条，效率高得多。

我在项目中遇到过一个问题：有个同事写了一个无状态的聚合逻辑，每次来数据都从 Kafka 重新拉取全量数据做计算。结果数据量一大，延迟直接飙到分钟级。后来改成有状态计算，延迟降到秒级。所以，选对状态模型，性能差距是数量级的。

1.2 状态的分类：Keyed State vs Operator State

Flink 把状态分成两大类。这个分类不是随便分的，它直接决定了你的算子怎么扩展、怎么恢复。

对比维度	Keyed State	Operator State
绑定对象	每个 Key 一个状态实例	每个并行子任务一个状态实例
适用场景	按 Key 分组后的聚合、窗口	Source/Sink 的偏移量、连接器元数据
扩缩容行为	Key 自动重新分配	需要实现 redistribute 逻辑
常用数据结构	ValueState, ListState, MapState, ReducingState, AggregatingState	ListState, UnionListState, BroadcastState

Keyed State

Keyed State 是最常用的。你只要在 DataStream 上做了 keyBy()，每个 Key 就会拥有自己独立的状态空间。我习惯把它理解成「每个 Key 的小仓库」。

举个例子：统计每个用户的登录次数。用户 A 来了，你更新 A 的计数器；用户 B 来了，你更新 B 的计数器。互不干扰。Flink 底层通过 Key Group 把状态分布到各个 TaskManager 上，扩缩容时自动做 Rebalance。

// 一个简单的 Keyed State 示例
public class LoginCountFunction extends RichFlatMapFunction<LoginEvent, Long> {
    private transient ValueState<Long> countState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> descriptor = 
            new ValueStateDescriptor<>("login-count", Types.LONG);
        countState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(LoginEvent event, Collector<Long> out) throws Exception {
        Long current = countState.value();
        if (current == null) {
            current = 0L;
        }
        countState.update(current + 1);
        out.collect(countState.value());
    }
}

Operator State

Operator State 是绑定到算子的每个并行实例上的。它不关心 Key，每个子任务维护一份独立的状态。

最常见的场景是 Kafka Source 的偏移量管理。每个 Source 子任务消费一个或多个分区，它需要记住自己消费到哪个位置了。这个偏移量就是 Operator State。

Operator State 的扩缩容比 Keyed State 麻烦。因为 Keyed State 可以按 Key Group 自动分配，但 Operator State 需要你实现 CheckpointedFunction 接口，自己写 redistribute 逻辑。

// Operator State 示例：记录每个子任务处理了多少条数据
public class MySourceFunction extends RichParallelSourceFunction<String> 
    implements CheckpointedFunction {
    
    private transient ListState<Long> offsetState;
    private long offset = 0L;

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Long> descriptor = 
            new ListStateDescriptor<>("offset", Types.LONG);
        offsetState = context.getOperatorStateStore().getListState(descriptor);
        
        // 从 Checkpoint 恢复
        if (context.isRestored()) {
            for (Long value : offsetState.get()) {
                offset = value;
            }
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        offsetState.clear();
        offsetState.add(offset);
    }
}

1.3 状态存储后端简介

状态存哪儿？Flink 提供了三种选择。我按推荐程度排个序：

1. HashMapStateBackend —— 存内存，适合小状态、高性能场景
2. EmbeddedRocksDBStateBackend —— 存本地磁盘，适合大状态、需要增量 Checkpoint 的场景
3. MemoryStateBackend —— 已废弃，别用了

怎么选？我一般这么判断：

• 状态小于 1GB，用 HashMap。性能最好，延迟最低。
• 状态大于 1GB，或者你不想让 JVM OOM，用 RocksDB。它把数据序列化后存到本地磁盘，内存只做缓存。
• 需要增量 Checkpoint？必须用 RocksDB。HashMap 只支持全量快照。

特性	HashMapStateBackend	EmbeddedRocksDBStateBackend
存储位置	JVM 堆内存	本地磁盘（RocksDB）
状态大小限制	受限于堆内存	受限于磁盘容量
读写性能	纳秒级	微秒级（有序列化开销）
增量 Checkpoint	不支持	支持
适用场景	小状态、低延迟	大状态、高可靠

1.4 本章知识体系总览

下面这张图把本章的核心知识点串起来了。你可以把它当作后续学习的导航图。