📚 课程目录第2章 Kafka Streams入门

2. Kafka Streams入门：核心概念与Hello World

说实话，我第一次接触Kafka Streams时，心里是有点抵触的。那时候我已经用Flink写了不少流任务，总觉得「又来个新框架，学不动了」。但真正上手之后，我发现它其实没那么复杂——甚至可以说，它是我见过最「轻」的流处理框架之一。

这一章，咱们就把它拆开揉碎了讲。你不需要有任何流处理基础，只要会用Java、知道Kafka的基本概念就行。

2.1 Kafka Streams到底是什么？

官方定义说：Kafka Streams是一个用于构建流处理应用的客户端库。注意「客户端库」这三个字——它不是一个集群，不是一个服务，就是一个jar包。

你想想看，这意味着什么？

你不需要搭建额外的集群（不像Flink要起JobManager/TaskManager）
你不需要运维任何流处理平台
你只需要写一个普通的Java应用，然后跑起来就行

我在项目中遇到过一位同事，他死活不肯用Flink，就是因为公司运维团队不给搭集群。后来我给他推荐了Kafka Streams，他花了一个下午就把原型跑通了。嗯，这就是它的魅力。

一句话总结：Kafka Streams = 一个能让你用Java写流处理逻辑的库，数据从Kafka进，处理完再写回Kafka。

2.2 三个核心概念

要理解Kafka Streams，你只需要抓住三个东西：拓扑、处理器、状态存储。其他的都是锦上添花。

2.2.1 拓扑（Topology）

拓扑就是你的数据处理流程图。它由节点（处理器）和边（数据流）组成。你写代码的时候，其实就是在构建一个有向无环图（DAG）。

我个人习惯把拓扑想象成「数据管道」：数据从源头Topic流进来，经过一系列处理节点，最后流到目标Topic。

2.2.2 处理器（Processor）

处理器是拓扑里的每个节点。它负责干一件事：拿到一条数据，做点什么，然后发出去。

Kafka Streams提供了两种处理器：

Processor API：底层API，你手动实现process()方法，想怎么处理就怎么处理
DSL API：高层API，用map()、filter()、groupBy()这些操作符，像写Java Stream一样写流处理

我建议新手先从DSL入手。为什么？因为代码量少，容易理解。等你需要做一些特殊操作（比如自定义状态管理）时，再回头用Processor API。

2.2.3 状态存储（State Store）

这个是最容易让人懵的概念。说白了，状态存储就是「内存里的一块数据」，用来记住之前处理过的信息。

举个例子：你要统计每个用户过去5分钟的点击次数。如果没有状态存储，你每次只能看到当前这一条数据，根本算不出「过去5分钟」的聚合结果。

Kafka Streams内置了RocksDB作为默认的状态存储引擎。它把数据写在本地磁盘上，但读写速度接近内存。我在项目中遇到过状态存储撑爆磁盘的情况——后来发现是key设计不合理，导致状态膨胀。这个坑后面会细说。

我的建议：刚开始学的时候，先别纠结状态存储的底层原理。你只需要知道「有状态的操作需要状态存储」就够了。等用到窗口聚合、join这些功能时，再回来深挖。

2.3 环境搭建

搭建Kafka Streams的开发环境，比你想象中简单。不需要装任何额外的软件，只要你有：

JDK 8+（我推荐JDK 11，稳定）
Maven或Gradle
一个本地Kafka集群（单节点就够了）

咱们用Maven来建项目。在pom.xml里加上这个依赖：

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

就这一个依赖，没有别的。你不需要引入kafka-clients，因为kafka-streams已经把它带进来了。

注意：版本号要和你本地Kafka集群的版本匹配。我曾经因为版本不匹配，折腾了半天才发现是序列化问题。Kafka 3.x 推荐用 3.6.0 或以上版本。

2.4 Hello World：单词计数

好了，咱们直接写代码。经典的WordCount，用Kafka Streams实现。

先看完整代码，我再一行行解释：

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.KTable;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Arrays;
import java.util.Properties;

public class WordCountApp {

    public static void main(String[] args) {
        // 1. 配置
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-app");
        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());

        // 2. 构建拓扑
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> textLines = builder.stream("input-topic");

        KTable<String, Long> wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word)
            .count();

        wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

        // 3. 启动
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // 优雅关闭
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

这段代码干了什么？我拆成三步来讲：

配置：设置应用ID（集群内唯一）、Kafka地址、序列化方式。这里用了String序列化，因为输入输出都是文本。
构建拓扑：从input-topic读数据，按空格分词，转小写，按单词分组，计数，结果写到output-topic。
启动：把拓扑和配置交给KafkaStreams，调用start()就完事了。

你发现没有？整个逻辑和Java Stream API几乎一模一样。区别只是数据源变成了Kafka Topic。

一个小技巧：调试的时候，可以在本地起一个控制台消费者，看看output-topic里到底写了什么：kafka-console-consumer --bootstrap-server localhost:9092 --topic output-topic --from-beginning --property print.key=true

2.5 拓扑可视化

上面那个WordCount的拓扑，我画了一张图，帮你直观理解数据是怎么流的：

从图上你能看到：数据从input-topic进来，经过分词、分组，然后count()操作会用到状态存储来维护每个单词的当前计数，最后结果写到output-topic。

2.6 跑起来看看

代码写完了，怎么跑？

启动Zookeeper和Kafka（如果你用的是KRaft模式，只需要启动Kafka）
创建两个Topic：kafka-topics --create --topic input-topic --bootstrap-server localhost:9092 和 output-topic
运行WordCountApp的main方法
往input-topic里发几条数据：kafka-console-producer --topic input-topic --bootstrap-server localhost:9092
在output-topic里看结果

你发一句「hello world hello kafka」，输出应该是：

hello    2
world    1
kafka    1

嗯，就是这么简单。

我曾经踩过的坑：第一次跑的时候，我忘了创建output-topic。Kafka Streams默认不会自动创建输出Topic，结果应用启动后一直没输出，我还以为是代码写错了。排查了半天才发现是Topic没建。所以记得先建好所有Topic。

2.7 本章小结

这一章我们干了三件事：

理解了Kafka Streams的本质——一个轻量级的流处理客户端库
掌握了三个核心概念：拓扑（数据流图）、处理器（处理节点）、状态存储（记住中间结果）
写了一个能跑的WordCount程序

说实话，Kafka Streams的上手门槛比Flink低很多。你不需要理解什么「检查点」、「水位线」、「算子链」这些概念，只要会写Java Stream，就能写Kafka Streams。

但别高兴太早——这只是开始。后面我们会遇到更复杂的问题：状态存储怎么调优？拓扑怎么拆分？怎么处理乱序数据？这些才是真正考验功底的地方。