Kafka Streams 入门

这一课你将能做到

1. 区分「定时批处理 SQL」与「Kafka Streams 实时拓扑」各自适合的报表场景，并给出判断依据
2. 读懂并运行一段最小 Streams 应用：过滤 OrderPaid → 按 key 计数 → 写回输出 Topic
3. 解释 APPLICATION_ID、状态 store、窗口（tumbling）与普通 Consumer 的差别

为什么要学

前面的课都是「生产 / 消费」：消费者拿到消息做副作用。但真实系统还有一类需求是流上变换：过滤、转换、聚合、窗口统计，然后把结果写回另一个 Topic 或供查询——比如「每分钟 GMV」「近 5 分钟支付失败率」「各课程实时点击榜」。

用普通 Consumer 手写聚合也能做，但你要自己处理：状态存哪、重启怎么恢复、分区怎么并行、rebalance 怎么不丢状态。Kafka Streams 是 JVM 内嵌库：不必另起 Flink/Spark 集群，一个普通 Java 应用就能做 Topic→Topic 处理，状态与容错框架代管。

本站主站报表仍可用 MySQL 聚合（数据量不大时更简单）；本课的目标是建立图像——知道 Streams 适合什么、长什么样，别把「再写一个 Consumer」和「流处理」混为一谈。

人话：Streams = 在消息管道上装一台小型加工机床，边流边算，断电了机床里的半成品还能恢复。

先建立图像

批处理：每小时 Job → SQL GROUP BY orders → 报表表（延迟小时级）
流处理：每条 OrderPaid → 拓扑 filter/map/group/count → 状态/输出 Topic（延迟秒级）

Kafka Streams 应用（就是普通 JVM 进程）
  application.id  ≈ 一个 Consumer Group（组内分摊分区）
  本地 RocksDB 状态 store + changelog Topic 做容错
  输入 Topic ──► 拓扑（filter/map/groupBy/aggregate/join）──► 输出 Topic

窗口类型	含义	例子
Tumbling	固定长度、不重叠	每 1 分钟 GMV
Hopping/Sliding	按步长滑动、可重叠	近 5 分钟滚动活跃数

普通 Consumer：自己写 poll 循环 + 自己管聚合状态
Streams：声明变换拓扑，框架管分区分配、状态恢复、再均衡

跟着做

步骤 1：先写清需求一句话

例：「对 coderoadmap.order-events 中的 OrderPaid 事件，按店铺维度做累计支付计数。」

如果需求是「每天凌晨出昨天的报表」，在笔记里标明：MySQL 批处理可能更简单——不必为了用 Streams 而用 Streams。判断口诀：要「秒级新鲜度」或「持续滚动窗口」→ 流；要「日终快照」→ 批。

步骤 2：加依赖并写最小拓扑

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>3.9.0</version>
</dependency>

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import java.util.Properties;

public class OrderPaidCountApp {
    public static void main(String[] args) {
        Properties props = new Properties();
        // 类似 consumer group：改名 ≈ 新应用，状态不共享
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "coderoadmap-order-stats");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("coderoadmap.order-events");

        source.filter((key, value) -> value != null && value.contains("OrderPaid"))
              .selectKey((key, value) -> "ALL")   // 演示：全局一个 key（生产应按 shopId）
              .groupByKey()
              .count()
              .toStream()
              .mapValues(count -> Long.toString(count))
              .to("coderoadmap.order-paid-stats", Produced.with(Serdes.String(), Serdes.String()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

运行，再用 console consumer 验证输出 Topic：

docker exec -it kafka-lab /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic coderoadmap.order-paid-stats --from-beginning --property print.key=true

多灌几条 OrderPaid，应看到计数 1、2、3 递增输出。注：groupByKey 前必须保证 key 分布合理，selectKey 会触发数据重分区（内部自动写重分区 Topic）。

步骤 3：想清楚聚合 key

如果消息 key 全是 orderId，groupByKey().count() 算的是「每订单支付次数」，不是「全站 GMV」。要按店铺聚合，应把 JSON 解析出 shopId 再 selectKey((k,v) -> shopId(v))。笔记写一句：聚合 key 是业务问题，先定 key 再写拓扑。

步骤 4：加一个 1 分钟窗口（纸上或代码）

import org.apache.kafka.streams.kstream.TimeWindows;
import java.time.Duration;

source.groupByKey()
      .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(1), Duration.ofSeconds(30)))
      .count()
      .toStream()
      .map((windowedKey, count) -> windowedKey.key() + "@" + windowedKey.window().startTime())
      .to("coderoadmap.paid-per-minute");

画图：10:00–10:01 一个桶，桶关闭输出 count，10:01 重新计。写一句：窗口解决「无限流上如何切出有边界的聚合」；grace 是允许乱序数据迟到的时间。

步骤 5：对比「要不要 Streams」

需求	更合适
转发邮件、写审计	普通 Consumer / Spring Listener
近实时排行、风控计数、滑动窗口告警	Streams 或其他流引擎
日终对账、月度报表	Cron + SQL 批处理

步骤 6：APPLICATION_ID 改名思想实验

问：把 coderoadmap-order-stats 改成 ...-v2，旧的计数状态还会接上吗？

答方向：application.id 决定内部 Topic（changelog、重分区 Topic）的名字，改名 = 全新应用，状态从零开始；要迁移状态需显式计划。所以改拓扑前先想 application.id 策略。

讲清楚要点

1. 因为Streams 是库不是集群，所以随你的 Spring Boot 应用部署扩缩，运维成本远低于独立流处理集群；但状态与再均衡要按官方运维指南理解。
2. 因为application.id 标识协作组并命名内部 Topic，所以乱改名字等于丢弃旧状态视图；它同时就是 Consumer Group id，组内自动分摊输入分区。
3. 因为聚合需要状态，所以Streams 用本地 RocksDB store + changelog Topic：重启时从 changelog 回放恢复，比「内存 HashMap 自己攒」可靠得多。
4. 因为窗口给无界流切界，所以实时看板用 tumbling/hopping 窗口，而不是无限 count 到底；窗口结果也带时间戳写出。
5. 因为简单副作用消费不需要拓扑，所以「发邮件、写审计」继续用 Listener；别为了炫技把业务逻辑塞进流拓扑。
6. 因为主站权威在 MySQL，所以Streams 输出是衍生指标/投影，对账与修正以库为准——它回答「现在怎么样」，不裁决「账对不对」。

你可能会懵的地方

1. 「Streams 和普通 Consumer 有何不同？」 都读 Topic。Consumer 给你原始 poll 循环，一切自理；Streams 给你声明式拓扑，框架管状态、分区分配、容错与Exactly-once 选项。判断：要不要「有状态的变换」？要 → Streams；只做无状态副作用 → Consumer 足够。

2. 「和 Flink 比谁强？」 Flink 是独立流引擎：事件时间语义更完整、大状态、复杂窗口、跨源 join 更强，但要单独运维集群。Streams 轻量内嵌、贴 Kafka 生态，中小规模实时统计首选。面试答「按规模与运维能力选」。

3. 「count 存哪？重启丢不丢？」 本地 RocksDB + changelog Topic 双份。实例崩溃，新实例从 changelog 回放恢复状态；application.id 不变就能接上。状态数据也受 retention 影响需正确配置。

4. 「filter 用字符串 contains 可以上生产吗？」 不能。示例图省事；生产用结构化 Serde/JSON 字段判断 event 类型，且解析失败要有处理路径（过滤/死信）。

5. 「为什么我 selectKey 后消息变多/变慢？」 selectKey 改变了 key→分区映射，Streams 内部自动生成重分区 Topic（副本、log 皆占资源）。大流量下这是真实成本，设计 key 时尽量避免中途换 key。

错法对照

错法	问题	正解
日终报表硬上 Streams	延迟要求根本不需要流	Cron + SQL 批处理
发邮件逻辑塞进复杂拓扑	副作用与流处理耦合，重试/幂等难做	副作用用 Listener + 幂等
随便改 APPLICATION_ID 还期望接续计数	改名 = 新应用，状态归零	版本化命名 + 显式状态迁移计划
无 key/乱 key 做聚合，结果不对	聚合维度错位	先定业务聚合 key，再写拓扑
用内存 HashMap 手写有状态聚合	重启/迁移状态漂移	用 Streams store 或明确外置存储

卡点急救

1. 应用启动报 Topology ... unknown topic 或卡在 Starting：输入 Topic 不存在。先按安装课 kafka-topics.sh --create，或开启开发参数 KAFKA_STREAMS_... 自动创建（仅实验）。
2. 输出 Topic 计数不对：九成是 key 问题——打印 windowedKey.key() 与输入 key 对照；确认 selectKey/groupByKey 的顺序符合预期。
3. 重启后计数清零：application.id 是否被改过？changelog Topic 是否被删？Streams 状态恢复依赖它们，删除即归零。

面试常问

• Kafka Streams 的状态是怎么容错的？ 每个有状态算子的本地 RocksDB store，都对应一个 changelog Topic（压缩日志）。写本地状态的同时写 changelog；实例故障后新实例通过回放 changelog 恢复状态到故障前。standby replica 可在另一实例预热副本，减少恢复时间。这就是「把 Kafka 的日志可靠度借给本地状态」。

• Streams 和 Flink 怎么选？ Streams：JVM 内嵌库、零额外集群、贴 Kafka、适合中小规模与 Kafka 中心架构；Flink：独立引擎、事件时间/大状态/复杂窗口与 Exactly-once 更成熟、适合大规模复杂流计算。答需求规模与运维投入，而不是背 feature 表。

• 什么是 tumbling window？乱序数据怎么办？ Tumbling：按固定时长切分、互不重叠、桶关闭后输出。乱序：事件时间窗口 + grace period——桶关闭后延迟 grace 内到达的数据仍计入并更新结果，超过则丢弃（或进侧输出）。Kafka Streams 以事件时间为准（消息自带 timestamp），处理时按事件时间分桶。

自测三问

1. APPLICATION_ID 改名对状态 store 有何影响？为什么？
2. 设计：有 click-topic，统计每分钟各课程 slug 的点击数——窗口与 key 怎么定？
3. Cron + MySQL 与 Streams 做近实时排行：运维差异写两条。

小结一句话

Streams = JVM 内嵌的 Topic 进、变换出流计算库：拓扑声明变换、框架管状态与容错；实验建立图像即可，日终报表仍优先批处理。

延伸

• Kafka Streams 官方文档
• 下一课「监控与 lag」：生产/消费链路跑通了，怎么知道它「健康」——从 LAG 看积压。