Flink运行时架构：JobManager、TaskManager、Slot、并行度与状态的关系
说实话，很多初学者学Flink状态管理，一上来就盯着API看，结果越学越迷糊。为什么？因为没搞懂运行时架构。状态不是凭空存在的，它得有个“家”——这个家就是TaskManager里的Slot。今天我就带你把这几个核心角色捋清楚。
一、Flink集群的“三驾马车”
一个Flink集群跑起来，主要靠三个角色配合：JobManager、TaskManager、Client。咱们重点说前两个，因为状态管理跟它们关系最密切。
1. JobManager：大脑
JobManager是整个集群的“指挥官”。它负责接收作业、生成执行计划、协调检查点、故障恢复。说白了，它不干活，但它知道谁该干什么。
我在项目中遇到过一个问题：JobManager挂了，整个作业就废了。所以生产环境一定要配HA（高可用），用ZooKeeper做leader选举。别省这个配置，血的教训。
2. TaskManager：工人
TaskManager是真正干活的。它启动多个线程，每个线程处理一个子任务。状态就存在TaskManager的内存或磁盘里。
每个TaskManager会向JobManager注册，报告自己有多少个Slot。JobManager根据这些信息分配任务。
3. Slot：工位
Slot是TaskManager内部的资源隔离单元。一个TaskManager可以配置多个Slot，每个Slot占用固定的内存。
举个例子：你给TaskManager分配了8GB内存，设置4个Slot，那每个Slot就是2GB。这个2GB包括算子计算用的内存，也包括状态存储用的内存。
核心理解：Slot是资源的最小分配单位，也是状态的最小承载单位。一个Slot里跑一个子任务，这个子任务的状态就存在这个Slot对应的内存区域里。
二、并行度：到底开几个线程干活？
并行度决定了你的算子有几个并行实例。比如并行度设为3，那Source、Map、Sink这些算子就会各开3个线程，分布在不同的Slot上。
并行度可以在四个层面设置，优先级从低到高：
| 配置层级 | 设置方式 | 生效范围 |
|---|---|---|
| 算子级别 | .map(...).setParallelism(4) |
单个算子 |
| 执行环境级别 | env.setParallelism(4) |
整个作业 |
| 提交作业时 | -p 4 |
覆盖环境设置 |
| 配置文件 | parallelism.default: 4 |
全局默认 |
我个人的习惯是：在算子级别设置关键算子的并行度，其他用环境默认值。这样灵活，也好调优。
三、并行度与Slot的关系：一张图说清楚
很多人搞不清并行度和Slot到底怎么对应。我画了一张图，你看完就明白了。
你看，并行度4的作业，需要至少4个Slot来跑。如果集群只有3个Slot，那第4个子任务就得等着。这就是为什么我建议你配置Slot数量时，要大于等于作业的最大并行度。
四、状态到底存在哪里？
这个问题我问过很多面试者，答案五花八门。其实很简单：状态存在TaskManager的Slot里。
具体来说：
- 堆内存状态（HeapKeyedStateBackend）：存在JVM堆里，速度快，但受GC影响大。适合小状态。
- RocksDB状态（RocksDBStateBackend）：存在本地磁盘，RocksDB管理。适合大状态，但读写慢一点。
我个人的经验是：状态小于1GB用堆内存，大于1GB用RocksDB。别纠结，选错了后面调优很痛苦。
小技巧：如果你用RocksDB，记得给TaskManager多分配一些堆外内存。因为RocksDB本身不占JVM堆，它用堆外内存和磁盘。我曾经因为没配堆外内存，导致RocksDB频繁刷盘，性能惨不忍睹。
五、并行度变化时，状态怎么办？
这是生产环境经常遇到的问题。你想想看，业务量涨了，想扩大并行度，但状态已经存在原来的Slot里了。怎么迁移？
Flink支持两种模式：
| 模式 | 说明 | 适用场景 |
|---|---|---|
| 重新缩放（Rescale） | 状态按照Key Group重新分配 | KeyedStream，状态按Key分布 |
| 均匀分布（Rebalance） | 状态轮询分配给新算子 | 非KeyedStream，Operator State |
举个例子：原来并行度2，状态分布在2个Slot里。改成并行度4后，Flink会把每个Slot里的状态按Key Group拆成两份，分给4个Slot。这个过程叫状态重分布。
注意：状态重分布是有代价的。它会触发一次全量快照，然后恢复。如果你的状态很大（比如几百GB），这个过程可能持续几分钟。我建议你在业务低峰期做扩缩容操作。
六、一个完整的例子：从架构到状态
说了这么多，咱们看一段代码。这是一个简单的计数作业，统计每个用户的点击次数。
public class ClickCountJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置并行度为4
env.setParallelism(4);
// 开启检查点，每5秒一次
env.enableCheckpointing(5000);
DataStream<ClickEvent> clicks = env
.addSource(new ClickSource())
.name("click-source");
DataStream<UserCount> counts = clicks
.keyBy(event -> event.userId)
.map(new RichMapFunction<ClickEvent, UserCount>() {
// 声明ValueState，存储每个用户的点击次数
private ValueState<Long> countState;
@Override
public void open(Configuration parameters) {
ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>("count", Types.LONG);
countState = getRuntimeContext().getState(descriptor);
}
@Override
public UserCount map(ClickEvent event) throws Exception {
Long currentCount = countState.value();
if (currentCount == null) {
currentCount = 0L;
}
currentCount++;
countState.update(currentCount);
return new UserCount(event.userId, currentCount);
}
})
.name("count-map");
counts.addSink(new PrintSink<>()).name("print-sink");
env.execute("Click Count Job");
}
}这个作业跑起来后，架构是这样的：
- JobManager接收作业，生成执行计划，把任务分配给TaskManager。
- 4个并行度，需要至少4个Slot。每个Slot跑一个完整的算子链：Source → Map → Sink。
- 每个Map算子的Slot里，都有一个
countState。它只存储分配到该Slot的那些userId的计数。 - 如果某个userId的点击事件被路由到Slot 0，那它的状态就存在Slot 0里。下次这个userId再来事件，还是去Slot 0。
为什么会这样？因为keyBy保证了同一个Key的数据永远去同一个分区。这是状态一致性的基础。
七、避坑指南：我踩过的几个坑
做Flink这几年，我在状态和架构上栽过不少跟头。分享几个典型的：
- Slot数量配少了：我曾经把Slot数量配成和并行度一样，结果一个TaskManager挂了，整个作业就瘫了。后来我改成Slot数量 = 并行度 × 1.5，留出冗余。
- 状态太大导致OOM：堆内存状态默认无界，状态大了直接撑爆JVM。后来我改用RocksDB，并设置了
state.backend.rocksdb.memory.managed: true，让Flink自动管理内存。 - 并行度变化后状态丢失：有一次我改了并行度，忘了从保存点恢复，结果状态全丢了。记住：改并行度一定要指定
--fromSavepoint。
一句话总结：JobManager管调度，TaskManager管执行，Slot管资源，并行度管并发，状态管数据。它们的关系就像公司架构：CEO（JobManager）分配任务，员工（TaskManager）在工位（Slot）上干活，每个员工负责一部分客户（状态），客户越多（并行度大）就需要更多工位。
好了，这一章的内容就到这里。你理解了运行时架构，后面学状态API、检查点、保存点就会轻松很多。记住我这句话：状态管理的问题，十有八九是架构没搞懂。