状态计算基础：有状态 vs 无状态，状态的定义与分类
聊流批一体，绕不开一个核心概念——状态。
我刚开始接触流计算时，对“状态”的理解很模糊。觉得不就是存点数据嘛，有啥好讲的？后来在线上环境踩了几次坑，才明白状态设计的好坏，直接决定了作业能不能跑稳、跑快。
这一节，咱们把状态计算的基础打扎实。
有状态计算 vs 无状态计算
先看个最简单的例子。你写一个 map 函数，输入一条数据，输出一条数据。它不依赖任何历史记录，这就是无状态计算。
// 无状态：输入什么，输出什么
DataStream<String> stream = env.fromElements("a", "b", "c");
stream.map(value -> value.toUpperCase());但如果你要算“截止到当前，一共来了多少条数据”，那就需要记住之前的结果。这个“记住”的能力，就是有状态计算。
核心区别一句话：无状态计算不依赖过去，有状态计算需要“记东西”。
我个人习惯这样区分：
- 无状态：幂等、易扩展、失败后重启无负担。适合做过滤、格式转换、简单映射。
- 有状态：需要维护累加器、窗口聚合、双流 join、去重逻辑。这些场景，没状态根本玩不转。
你想想看，如果做实时去重，不记住之前来过哪些数据，怎么判断新来的数据是不是重复？
避坑指南：我曾经接手过一个项目，同事把所有状态都塞在内存 HashMap 里。作业跑了一周，一次重启后数据全乱了。为什么？因为 HashMap 的状态不会自动恢复。记住：生产环境一定要用框架管理的状态，别自己搞。
状态的定义与分类
Flink 把状态分成两大类：Keyed State 和 Operator State。
Keyed State（键控状态）
Keyed State 是“按 key 隔离”的状态。什么意思呢？就是每条数据根据它的 key，只访问自己那份状态。
举个例子：统计每个用户的点击次数。用户 A 的计数不会影响用户 B 的计数。这就是 Keyed State 的典型场景。
// Keyed State 示例：统计每个用户的点击次数
DataStream<ClickEvent> clicks = ...;
clicks
.keyBy(event -> event.getUserId())
.process(new KeyedProcessFunction<String, ClickEvent, Long>() {
private ValueState<Long> countState;
@Override
public void open(Configuration parameters) {
ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>("clickCount", Types.LONG);
countState = getRuntimeContext().getState(descriptor);
}
@Override
public void processElement(ClickEvent value, Context ctx, Collector<Long> out) {
Long current = countState.value();
if (current == null) current = 0L;
countState.update(current + 1);
out.collect(countState.value());
}
});Keyed State 有几种常见类型：
| 状态类型 | 说明 | 适用场景 |
|---|---|---|
| ValueState<T> | 单个值，类型由 T 指定 | 计数器、开关标志 |
| ListState<T> | 列表，可追加元素 | 缓存最近 N 条记录 |
| MapState<UK, UV> | 键值映射 | 维度表缓存、配置映射 |
| ReducingState<T> | 自动聚合，保留聚合结果 | 累加求和、求最大值 |
| AggregatingState<IN, OUT> | 自定义聚合逻辑 | 复杂指标计算 |
注意：Keyed State 必须在 keyBy 之后才能使用。如果你没做 keyBy，直接调用 getState()，运行时会报错。这个坑我见过不止一次。
Operator State（算子状态）
Operator State 是“算子级别”的状态。它不按 key 划分，而是按并行子任务划分。每个并行实例维护自己的一份状态。
典型的应用场景：Kafka 连接器的 offset 记录。每个并行子任务消费自己的分区，记录自己消费到的位置。
// Operator State 示例：记录每个子任务处理过的数据条数
public class MyStatefulSource extends RichParallelSourceFunction<Long> {
private ListState<Long> offsetState;
private long offset = 0;
@Override
public void open(Configuration config) {
ListStateDescriptor<Long> descriptor =
new ListStateDescriptor<>("offset", Types.LONG);
offsetState = getRuntimeContext().getListState(descriptor);
}
@Override
public void run(SourceContext<Long> ctx) {
// 从状态恢复 offset
for (Long value : offsetState.get()) {
offset = value;
}
// 生产数据...
}
}Operator State 目前主要支持两种：
- ListState：每个并行子任务维护一个列表。重启时，所有子任务的列表合并起来，再重新分配。
- UnionListState：类似 ListState，但重启时把完整列表广播给每个子任务。适合需要全局视图的场景。
状态后端（State Backend）选型
状态存哪儿？怎么存？这就是状态后端要解决的问题。
Flink 提供了三种状态后端，我按自己的使用经验给你排个序：
| 状态后端 | 存储位置 | 适用规模 | 我推荐的使用场景 |
|---|---|---|---|
| MemoryStateBackend | TaskManager 内存 + JobManager 内存（快照） | 极小（百 MB 以内） | 本地开发调试，千万别上生产 |
| FsStateBackend | TaskManager 内存 + 外部文件系统（快照） | 中等（GB 级别） | 小规模生产，状态不大时可用 |
| RocksDBStateBackend | RocksDB（本地磁盘） + 外部文件系统（快照） | 超大（TB 级别） | 生产首选，能扛大状态 |
我的建议：生产环境无脑选 RocksDB。除非你的状态真的只有几百 MB，而且对延迟极其敏感，才考虑 FsStateBackend。MemoryStateBackend 只适合在 IDE 里跑着玩。
为什么我力推 RocksDB？
- 它把状态存在本地磁盘，不受 JVM 堆内存限制。你想想看，如果状态有 100GB，堆内存怎么可能装得下？
- RocksDB 支持增量 checkpoint，只上传变化的部分，大大减少快照时间。
- 它天然支持异步快照，不会阻塞正常的数据处理。
但 RocksDB 也有代价：读写需要序列化/反序列化，比纯内存访问慢一些。不过对于绝大多数场景，这点延迟完全可以接受。
选型口诀：小状态用内存，大状态用 RocksDB。不确定大小时，直接上 RocksDB 最稳妥。
下面这张图总结了状态计算的核心知识体系：
嗯，这一节的内容就到这儿。状态计算的基础打牢了，后面讲状态复用、增量 checkpoint、容灾恢复时，你才能理解为什么有些设计是必要的。