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));
    }
}