实操：把原理写成一个能跑的 Streams 应用

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-streams</artifactId>
  <version>4.3.0</version>
</dependency>
<!-- 序列化常用 JSON 时再加（本章 worked example 只用内置 Serdes，可不加）：
<dependency>
  <groupId>com.fasterxml.jackson.core</groupId>
  <artifactId>jackson-databind</artifactId>
  <version>2.17.0</version>
</dependency> -->

# 输入 topic：单词计数的文本行（3 分区——决定了并行上限）
bin/kafka-topics.sh --create --topic text-lines \
  --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

# 输出 topic：可不预建（to() 写入时若 auto.create.topics.enable=true 会自动建），
# 但生产环境建议显式建，控制分区数与配置
bin/kafka-topics.sh --create --topic word-counts-output \
  --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

# 验证已建：
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

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.Topology;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

public class AppSkeleton {
    public static void main(String[] args) {
        Properties props = new Properties();
        // application.id 同时就是消费组 group.id —— 改名 = 一个全新消费组 = 状态从头来
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 默认 serde：没有在算子里显式给 Consumed/Produced/Materialized 时，用这两个兜底
        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();
        // …在这里用 builder 描述拓扑（见 §3.2）…
        Topology topology = builder.build();
        System.out.println(topology.describe());   // 打印拓扑：能看清 Streams 实际建了哪些内部 topic

        KafkaStreams streams = new KafkaStreams(topology, props);
        CountDownLatch latch = new CountDownLatch(1);
        // 优雅关停：close() 会提交 offset、刷写状态、退出消费组，避免下次启动多走一次恢复
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            streams.close();
            latch.countDown();
        }));

        streams.start();   // 此刻才真正加入消费组、开始拉取与处理
        try {
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;
import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

public class WordCountApp {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count-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());

        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> lines =
            builder.stream("text-lines", Consumed.with(Serdes.String(), Serdes.String()));

        KTable<String, Long> counts = lines
            // stateless：一行拆成多个词，值变了、key 还是原 key（此处不重分区）
            .flatMapValues(line -> Arrays.asList(line.toLowerCase(Locale.ROOT).split("\\W+")))
            // 改 key：把词本身设成 key —— 触发下游重分区（§2.3）
            .groupBy((key, word) -> word, Grouped.with(Serdes.String(), Serdes.String()))
            // stateful：计数，物化进名为 counts-store 的 RocksDB + 一条 changelog
            .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store")
                       .withKeySerde(Serdes.String())
                       .withValueSerde(Serdes.Long()));

        // KTable → 变更流 → 写回输出 topic（值是 Long，需显式 Long serde）
        counts.toStream()
              .to("word-counts-output", Produced.with(Serdes.String(), Serdes.Long()));

        Topology topology = builder.build();
        System.out.println(topology.describe());

        KafkaStreams streams = new KafkaStreams(topology, props);
        CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread(() -> { streams.close(); latch.countDown(); }));
        streams.start();
        try { latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

# 1) 启动应用（假设已 mvn package 出可执行 jar / 或在 IDE 里 run main）
java -cp target/streams-demo-1.0.jar WordCountApp

# 2) 终端 A：输入文本行
bin/kafka-console-producer.sh --topic text-lines --bootstrap-server localhost:9092
> kafka streams kafka
> streams streams

# 3) 终端 B：读输出（值是 Long，要指定反序列化器）
bin/kafka-console-consumer.sh --topic word-counts-output \
  --bootstrap-server localhost:9092 --from-beginning \
  --property print.key=true --property key.separator=" => " \
  --value-deserializer org.apache.kafka.common.serialization.LongDeserializer

# 灌入 "kafka streams kafka" 后：
kafka   => 1
streams => 1
kafka   => 2
# 灌入 "streams streams" 后：
streams => 2
streams => 3

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Locale;

StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> lines =
    builder.stream("text-lines", Consumed.with(Serdes.String(), Serdes.String()));

KTable<Windowed<String>, Long> windowedCounts = lines
    .flatMapValues(line -> Arrays.asList(line.toLowerCase(Locale.ROOT).split("\\W+")))

    // ── 决策点 1：groupByKey 还是 groupBy？──────────────────────────
    // flatMapValues 没改 key（key 仍是原始行 key），但聚合要按「词」分组。
    // TODO: 选 groupByKey() 还是 groupBy((k, word) -> word, ...)?
    .???(/* ... */)

    // ── 决策点 2：窗口类型 + grace 宽限期 ──────────────────────────
    // 需求是「每 5 分钟一个不重叠的窗口」。
    // TODO: 用哪种 TimeWindows？要不要 grace？grace 给多久？
    .windowedBy(/* TODO */)

    // ── 决策点 3：Materialized 要不要显式 serde？──────────────────
    // 聚合值类型是 Long。
    // TODO: 这里能只写 Materialized.as("win-counts") 吗？还是必须补 serde？
    .count(/* TODO */);

windowedCounts.toStream()
    // Windowed<String> 的 key 需要 WindowedSerdes，值是 Long
    .to("windowed-counts-output",
        Produced.with(WindowedSerdes.timeWindowedSerdeFrom(String.class, 5 * 60 * 1000L),
                      Serdes.Long()));