什么是反压
反压（Backpressure）是流处理系统的「自我保护机制」。当下游算子处理速度跟不上上游时，系统通过阻塞或丢弃数据来避免自身崩溃。
快速 Source → 慢速 Window 聚合 → 更慢的 Redis Sink
↑
Window 缓存区满 → 阻塞上游 Map
↑
Map 缓存区满 → 阻塞 Source
↑
Source 停止从 Kafka 拉取数据
反压不是故障，而是系统在过载时的正常反馈。持续、大面积的反压才说明处理能力不足；完全不出现反压通常是好事，说明当前负载在设计容量之内。
Web UI 中的反压指示
Flink Web UI (localhost:8081) 的 Backpressure 标签页显示每个算子的反压比例：
| 状态 | 颜色 | 含义 |
|---|---|---|
| OK | 绿色 | 反压比例 ≤ 10%（正常范围） |
| LOW | 橙色 | 10% < 比例 ≤ 50% |
| HIGH | 红色 | 比例 > 50% |
检测方法不是看缓冲区水位。Flink 的 Backpressure 采样器检查该 Task 的输出分区是否还可写：不可写的时长占比就是反压比例，对应指标是 backPressuredTimeMsPerSecond（毫秒/秒）。它衡量的是「输出被下游堵住」的时间，不是「线程在忙」的时间。
定位瓶颈
反压表示「我的输出写不出去」，所以它从下游往上游传：
Source(OK) → Map(LOW) → KeyBy(HIGH) → Window(HIGH) → Sink(OK)
↑
Sink 慢，Window 的输出先堵住，
再把 KeyBy、Map 依次顶成反压
Sink 显示 OK 不代表它没慢：Sink 没有输出分区，通常永远不会显示反压。上图的瓶颈是 Sink，不是 KeyBy 或 Window——它们只是被下游堵住后把压力继续往上游传。
定位方法：从 Source 往下游方向找，最后一个非 OK 的算子，它的下游才是瓶颈。 如果所有上游算子都显示反压而 Sink 是 OK，优先怀疑 Sink 或 Sink 写入的外部系统；再结合 busyTimeMsPerSecond、records-in / records-out 和外部系统指标确认。
反压的三种处理策略
| 策略 | 何时用 | 怎么做 |
|---|---|---|
| 扩容 | 算子逻辑本身不重，而是并行度不够 | 增加并行度（或增加 TaskManager Slot） |
| 优化 | 算子逻辑太复杂 | 减少状态访问、避免全量聚合、使用增量计算 |
| 降级 | 允许精度损失、不允许丢数据 | 采样、合并维度、跳过非关键 enrichment |
Flink 没有「配置一下就开始丢数据」的开关：buffer-timeout 只影响输出批次的大小和延迟，侧输出也不是丢弃。要真的丢，只能在业务代码里显式跳过，并且必须把这个决定写进验收标准。
Network Buffer 参数
Flink 在 TaskManager 之间传输数据用网络缓冲（Network Buffer）。参数配置在 flink-conf.yaml：
# 每个 TaskManager 的网络缓冲区数量
taskmanager.memory.network.fraction: 0.15
taskmanager.memory.network.min: 128mb
taskmanager.memory.network.max: 512mb
# 单个内存段大小（影响缓冲粒度）
taskmanager.memory.segment-size: 32kb
反压与缓冲的关系
小缓冲（少量 Segment）→ 反压更敏感 → 更快触发反压 → 更少的缓存延迟
大缓冲（大量 Segment）→ 反压更迟钝 → 更多数据堆积 → 更高的延迟但吞吐量更大
eventd v3 的配置偏向低延迟：
# 低延迟场景：减少缓冲，让反压更快传播
taskmanager.memory.network.fraction: 0.1
taskmanager.memory.network.min: 64mb
如果作业吞吐量为主而非延迟，可以增加缓冲减少反压：
# 高吞吐场景：增加缓冲
taskmanager.memory.network.fraction: 0.2
taskmanager.memory.network.min: 256mb
Slot 共享与资源分配
默认 Slot 共享
Flink 默认允许不同算子共享同一个 Slot。这意味着一个 4 并行度的作业，如果 Source 和 Sink 都共享 Slot，最小只需要 4 个 Slot 即可运行。
Slot 1: Source(1) → Map(1) → Sink(1)
Slot 2: Source(2) → Map(2) → Sink(2)
Slot 3: Source(3) → Map(3) → Sink(3)
Slot 4: Source(4) → Map(4) → Sink(4)
优点：节省资源，减少数据传输开销（同一 Slot 内直接内存传递）。
缺点：如果某个算子特别慢，会拖慢整个 Slot 内所有算子。
设置 Slot 共享组
用 slotSharingGroup 隔离重算子：
DataStream<Result> result = stream
.map(new LightTransform())
.slotSharingGroup("light") // 轻量算子组
.keyBy(Event::getUserId)
.window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
.aggregate(new HeavyAggregate())
.slotSharingGroup("heavy") // 重量算子组
.map(new OutputTransform())
.slotSharingGroup("light");
“heavy” 组的算子不与其他组共享 Slot。这会占用更多 Slot，但避免重算子拖慢轻量算子。
并行度调优
Source 并行度
Source 并行度通常等于上游分区数：
// Kafka Source 并行度 = Kafka Topic 分区数
KafkaSource<Event> source = KafkaSource.<Event>builder()
.setTopics("eventd-cleaned")
.setBootstrapServers("kafka:9092")
// 不设反序列化器时 build() 会直接抛 "Deserialization schema is required"
.setValueOnlyDeserializer(new EventDeserializationSchema()) // 定义见 7.1
.build();
env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka")
.setParallelism(12); // 12 个 Kafka 分区 → 并行度 12
Sink 并行度
Sink 通常受限于外部系统的吞吐：
// Redis 单实例 QPS 有限，Sink 并行度按连接池与命令往返时间定
stream.addSink(new RedisMetricSink()) // 自定义 RichSinkFunction，见 7.2
.setParallelism(4); // 4 个并行子任务，各自持有一条连接
// Kafka Sink 可以和 Source 一致
stream.sinkTo(kafkaSink)
.setParallelism(12);
中间算子并行度
中间算子（map、filter、flatMap）通常跟随上游：
// 保持链式执行，不单独设置并行度
stream.map(...).filter(...).keyBy(...)
如果中间算子特别重，可以增加并行度：
stream.map(new ExpensiveComputation())
.setParallelism(8); // 比上游多 8 个实例
内存调优
Flink 1.10+ 使用统一内存管理，TaskManager 内存分为：
| 区域 | 配置参数 | 说明 |
|---|---|---|
| Framework Heap | taskmanager.memory.framework.heap.size | Flink 框架自身 |
| Task Heap | taskmanager.memory.task.heap.size | 用户代码 |
| Managed Memory | taskmanager.memory.managed.size / .fraction | 批算子、RocksDB 状态 |
| Network | taskmanager.memory.network.min / .max / .fraction | 网络缓冲 |
| JVM Overhead | taskmanager.memory.jvm-overhead.min / .max / .fraction | JVM 自身开销 |
taskmanager.memory.process.size 是上面各项的总预算，不是单独一块内存。
RocksDB 状态内存
使用 EmbeddedRocksDBStateBackend 时，状态数据落在本地磁盘（state.backend.rocksdb.localdir），托管内存只用来放 block cache、write buffer 这些 RocksDB 的内存结构；托管内存给少了，RocksDB 会频繁读盘：
taskmanager.memory.managed.fraction: 0.4 # 40% 给托管内存（RocksDB）
如果状态很大：
# 大状态场景
taskmanager.memory.managed.size: 2g # 显式分配 2GB
state.backend.incremental: true # 增量 Checkpoint
反压排查清单
eventd v3 的调优检查点：
- 反压位置：Web UI 中哪个算子 HIGH？
- 是否扩容瓶颈：并行度是否小于上游分区数？
- 状态大小：Checkpoint UI 中状态数据量？
- 垃圾回收：GC 暂停是否明显偏离自己的基线（不要用固定的 100ms 阈值）
- 网络缓冲：是否配置过小导致频繁反压？
- KeyBy 热点：某个 key 数据量是否远大于其他？
KeyBy 热点排查
stream.keyBy(Event::getUserId)
.process(new KeyedProcessFunction<String, Event, Event>() {
private transient Counter recordCounter;
private transient Logger log;
@Override
public void open(OpenContext openContext) {
log = LoggerFactory.getLogger(getClass());
// 反例是 counter("keyCount_" + key)：key 有多少种就注册多少个
// 指标，直接把 Metrics 注册表撑爆。这里用固定名字
recordCounter = getRuntimeContext()
.getMetricGroup()
.counter("records-processed");
}
@Override
public void processElement(Event event, Context ctx, Collector<Event> out) {
recordCounter.inc();
// 每 10 万条采样打一次日志，人工确认是不是同一个 key 在刷
if (recordCounter.getCount() % 100_000 == 0) {
log.info("hot key candidate: {}", ctx.getCurrentKey());
}
out.collect(event);
}
});
更系统的做法是对 key 做有界的 heavy-hitter 统计（如 Count-Min Sketch），而不是给每个 key 建一个指标——key 的基数由业务决定，指标数量不能跟着它涨。
如果某个用户的计数远大于其他，说明该 key 是热点。解决方案：
- 对热点 key 做进一步分区（如按用户上的二级 key）
- 将该 key 的数据分散到多个子任务
事件时间延迟 vs 反压
两个不同的问题，容易混淆：
| 现象 | 根本原因 | 指标 |
|---|---|---|
| Watermark 滞后 | 数据到达慢 | currentOutputWatermark 远小于系统时间 |
| 反压 | 算子处理慢 | backPressuredTimeMsPerSecond > 0 |
Watermark 滞后 说明上游数据来得慢（或者某个分区 idle）。这不是系统性能问题，而是数据生产速率问题。
反压 说明数据到达正常，但系统处理不完。这是系统性能问题。
小结
| 概念 | 说明 |
|---|---|
| 反压 | 下游慢 → 上游阻塞 |
| 瓶颈定位 | Web UI 从后往前找第一个非 OK 的算子 |
| Network Buffer | 大小的反压敏感度 |
| Slot 共享组 | 隔离重算子 |
| 并行度 | Source = 分区数，Sink = 外部系统能力 |
| 托管内存 | RocksDB 状态存储 |
下一节 8.3 测试：MiniCluster 与测试工具 ：MiniCluster、TestStreamEnvironment、CollectSink。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。