14.1 性能基线与调优

先把「性能」拆成四个口径

「吞吐 10 万条/秒」「延迟 3 秒」这类说法，如果没有说明在哪一层测的，基本无法比较。流处理至少涉及四个不同口径：

Flink 的 LatencyMarker 测的是作业内部从 Source 到 Sink 的传递时间，不包含 Kafka 生产、消费和下游写入。它适合比较两个作业版本的相对快慢，不能直接当 SLO。

基线必须绑定环境

性能基线不是一组常数，而是「这套代码 + 这套配置 + 这台机器 + 这份数据」的组合。记录模板：

三条纪律：

1. 同一 commit、同一份数据、同一时钟。三次测量取中位数，丢掉启动后的预热段。
2. 报告分位数。平均值会掩盖长尾；至少给 p50、p95、p99。
3. 不要抄别人的数字。换了机器或分区数，结论就可能反过来。

吞吐怎么测

先用有界源测「这套算子的上限」：

public class ThroughputBenchmark {
    record BenchEvent(long id, String userId, double amount, long eventTime) {}

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        long total = 10_000_000L;

        DataStream<Long> source = env.fromSequence(1, total);
        source.map(i -> new BenchEvent(i, "u" + (i % 100), 1.0, System.currentTimeMillis()))
              .addSink(new DiscardingSink<>());

        long start = System.nanoTime();
        env.execute("throughput-benchmark");
        double seconds = (System.nanoTime() - start) / 1e9;
        System.out.printf("throughput = %.0f records/s%n", total / seconds);
    }
}

这段代码测的是纯处理上限：没有 Kafka 网络、没有反序列化、没有下游写入。它能回答「瓶颈在不在算子」，不能回答「线上能扛多少」。另外分母包含了作业启动与序列化时间，千万级数据下可忽略，数据量小的时候必须先跑一轮预热、再计时第二轮。

要测持续吞吐，必须让输入按目标速率持续一段时间，并同时观察：

• Sink 的 numRecordsInPerSecond 是否跟得上 Source；
• backPressuredTimeMsPerSecond 是否持续升高；
• Kafka lag 是否单调增长；
• Checkpoint duration 是否被拖长。

分别跑「关闭 Checkpoint」和「开启 Checkpoint」两组。两组差异就是容错开销，不要混在一起报一个数字。

延迟怎么测

先区分两种延迟：

// 事件时间滞后：数据有多旧。重放历史数据时天然很大
long dataAgeMs = System.currentTimeMillis() - event.eventTime();

// 端到端延迟：从生产到结果可见。
// v3 的 Event 没有 producedAt 字段，生产端把它写进 metadata 扩展字段
public class EndToEndLatency extends RichMapFunction<Event, Event> {
    private transient Histogram histogram;

    @Override
    public void open(OpenContext ctx) {
        histogram = getRuntimeContext().getMetricGroup()
            .histogram("endToEndLatencyMs", new DescriptiveStatisticsHistogram(4096));
    }

    @Override
    public Event map(Event event) {
        String producedAt = event.metadata() == null
            ? null : event.metadata().get("producedAt");
        if (producedAt != null) {
            histogram.update(System.currentTimeMillis() - Long.parseLong(producedAt));
        }
        return event;
    }
}

DescriptiveStatisticsHistogram 来自 flink-runtime，容量按窗口内样本量设置；样本溢出会丢尾部数据，容量不要拍脑袋。指标导出后看 p50/p95/p99，不要只看平均。

如果 Sink 是 Redis 这类「写完即返回」的系统，端到端延迟还应包含写入耗时；只测到 map 结束会低估。

Checkpoint 间隔怎么选

这张表里的「重放窗口」只描述状态恢复到哪个时间点，不等于端到端「最多丢 10 秒数据」。故障后是否丢、是否重，取决于：

• Source 是否在 Checkpoint 完成时提交 offset；
• Sink 是否事务写或幂等写；
• 下游是否按业务键去重。

把 Checkpoint 间隔直接写成「最坏丢数据时长」，是把三个层面的问题混成了一个。

工程上常用的经验规则是 Checkpoint duration < interval 的 50%。超过这条线不必然失败，但说明快照已经接近超时，遇到反压或存储抖动就容易连续失败。反压严重时，对齐阶段会进一步拉长 duration，可以考虑非对齐 Checkpoint，代价是快照体积变大。

并行度扩展性怎么测

不要从「几核就设几」开始，先做实验：

1. 固定输入速率，确保它低于当前配置的处理上限；
2. 并行度分别取 1、2、4、8，其他配置不变；
3. 每组跑够时间，记录吞吐、p99 延迟、反压时间和 Checkpoint duration；
4. 找到吞吐不再随并行度线性增长的拐点；
5. 检查拐点处哪个算子先饱和（Source / 窗口 / Sink）；
6. 如果 Source 并行度受 Kafka 分区数限制，先加分区，否则加并行度只是多出空闲 Source 子任务。

扩展性受四个因素约束：

• Kafka 分区数：Source 并行度不能超过分区数；
• key 倾斜：某个 userId 占大部分流量时，加并行度也只有一个子任务忙；
• Sink 吞吐：Redis/数据库连接池往往先于 Flink 饱和；
• 状态后端：RocksDB 的 I/O 和锁竞争会在某个并行度后显现。

报告里要给出自己的曲线和拐点，而不是引用一个「并行度 8 提升 1.63 倍」的通用数字。

Kafka Streams 的测量

TopologyTestDriver 是正确性工具，不是吞吐基准。它在单线程内存里驱动拓扑，没有 broker、没有网络、没有 Consumer fetch 和 Producer batching；跑出来的「条/秒」不能当线上容量。

测 Kafka Streams 持续吞吐，用外部工具和 JMX：

kafka-producer-perf-test.sh --topic eventd-raw \
  --num-records 1000000 --record-size 512 --throughput -1 \
  --producer-props bootstrap.servers=localhost:9092

kafka-consumer-perf-test.sh --topic eventd-cleaned \
  --messages 1000000 --bootstrap-server localhost:9092

应用侧看 JMX 的 process-rate、commit-rate、poll-latency-avg，以及 Consumer Group 的 lag。优化时优先动这些参数：

// 3.8 起的正确名字；cache.max.bytes.buffering 已废弃
props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 10 * 1024 * 1024L);

props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024 * 1024);
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 100);

如果开启 EOS，还要注意 commit.interval.ms 与 transaction.timeout.ms 的硬约束：EOS 下 commit 间隔默认 100ms、事务超时默认 10s，事务超时必须不小于 commit 间隔，否则 new KafkaStreams(...) 会直接抛 IllegalArgumentException（见 3.2）。

反压与降级

先定位，再降级。反压的入口指标是 backPressuredTimeMsPerSecond、输出缓冲池使用率和下游 Sink 延迟；Kafka lag 持续增长只是结果，不是原因。

有效的杠杆：

不是杠杆的操作：

• 调大 Checkpoint 间隔：它降低快照开销，但不解决处理不过来的问题，只是让积压更晚被发现。
• 调大 allowedLateness：它让窗口保留更久、接收更多迟到数据，反而增加状态和重复触发。把它当成「减少重算」是一种误读。
• 把 Event Time 换成 Processing Time：结果不再可复现，故障恢复后对不上账。
• 盲目缩小窗口：窗口变小不会让单条记录更便宜，反而增加触发次数和下游写入压力。

调优检查清单

对着基线看，而不是对着固定阈值看：

□ Source 并行度 ≤ Kafka 分区数
□ 每个算子的 in/out 速率与基线偏差在可解释范围内
□ backPressuredTimeMsPerSecond 未持续升高
□ Checkpoint duration < interval 的 50%
□ Checkpoint 对齐时间没有随负载显著上升
□ Kafka lag 在生产停止后能回落到基线
□ GC 暂停与基线相比没有数量级变化
□ 状态大小增长有上界（TTL 或窗口清理生效）
□ Redis/数据库连接池等待时间稳定

「GC < 100ms」「SST 文件 < 100」这类固定阈值只能当报警线，不能当调优目标；换一台机器就不成立。

小结

下一节 14.2 故障排查 ：反压、状态过大、OOM、Watermark 滞后。