核心概念：Streams 是跑在消费组上的库

读这章前要带着主 Kafka 教程的三块底座：分区是并行单位、消费组按分区分活、offset 记录消费进度。Streams 不替换这些——它把这些当地基，往上盖了一层"有状态消费 + 流处理 DSL"。下面六节按依赖顺序展开，每节末尾点出它和下一个概念的接口。

1.1为什么用 Streams 而非裸 consumer

Streams 把"有状态消费"里那几件难做对的事——状态存哪、怎么容错、再平衡后怎么不重算——收进库里替你做。

这三件事不是边角料，是有状态流处理的全部难点。Streams 的设计是把它们各给一个机制：状态进本地 state store（默认 RocksDB，§2.1 详解）；容错靠把每次状态写也写进一条 changelog topic，崩溃后回放重建；不重算靠复用消费组的 offset 语义，并把 changelog 写、offset 提交、输出 produce 三者在 EOS 下绑进一个事务（§2.6 详解）。

场景走查：从裸 consumer 到一个 Streams 拓扑

同一个"实时下单计数"，用 Streams 写出来是这样——状态、容错、再平衡那三件事一行都看不到，因为它们被库吞了：

StreamsBuilder b = new StreamsBuilder();
b.stream("orders", Consumed.with(Serdes.String(), orderSerde))  // KStream：每笔订单一条
 .groupBy((k, order) -> order.userId())                          // 按用户重新分组
 .count(Materialized.as("order-counts"))                         // KTable：每用户当前计数
 .toStream().to("order-counts-out");                             // 变更流写回 topic

四行代码背后，是这一整章要拆开的机制。与下一个概念的关系：第 1 行的 stream(...) 产出一个 KStream，第 3 行的 count() 产出一个 KTable——这两个类型是 Streams 全部 DSL 的两块基石，下一节把它们和 GlobalKTable 一起讲清楚。

1.2KStream / KTable / GlobalKTable

KStream 是只追加的事件流（每条都留），KTable 是按 key 取最新值的变更表，GlobalKTable 是每个实例都持有全部分区的只读副本。

底层机制（比文档深一层）

KStream = INSERT 语义。每条记录都是独立事实，相同 key 的多条记录全部保留、依次处理。对 KStream 做 count，结果随每条记录单调增长。

KTable = UPSERT 语义，本质是一条 changelog。记录按 key 折叠：相同 key 的新记录覆盖旧值，下游只看到"每 key 当前值"。关键的一条是 value == null 是 tombstone（墓碑）——它表示"这个 key 被删除"，而不是"这个 key 的值是空"。KTable 把上游 topic 当成 compacted 日志来读：日志里同一 key 留最新、tombstone 触发删除，正是 log compaction 的语义。这就是为什么文档说"KTable 是 changelog stream 的视图"——它读的就是一条 changelog。

GlobalKTable = 每实例全分区副本。普通 KTable 每个实例只持有它被分到的那些分区；GlobalKTable 让每个实例在启动时 eager 加载该 topic 的所有分区，得到一份完整副本。代价是每实例都扛全量数据（磁盘 + 内存 + 重启时的 bootstrap 延迟），所以它只配小而慢变的维表用（国家码、商品目录）。回报是它免去共分区要求——join 时流的 key 不必和表的分区方式对齐（§2.5）。

场景走查：同一条 topic，两种读法

topic user-tier 依次到达三条记录：(alice, silver)、(bob, gold)、(alice, gold)。

• 读成 KStream：下游看到 三 条事件，依次 silver、gold、gold。它把"alice 升级"当成一次发生过的事实记下来。
• 读成 KTable：下游看到的表是 {alice: gold, bob: gold}——alice 的 silver 被第三条覆盖。再来一条 (alice, null)，表变成 {bob: gold}，alice 这个 key 被 tombstone 删掉。

KStream<String, String> tierEvents = b.stream("user-tier");  // 三条事件全留
KTable<String, String>  tierNow    = b.table("user-tier");   // 每用户最新等级
GlobalKTable<String, String> country = b.globalTable("country-codes"); // 每实例全量副本

与下一个概念的关系：上面这道题里"KTable 对外发的是变更日志、表是变更折叠的结果"——把这句话正着读和反着读，就是下一节的流表对偶。

1.3流表对偶

表是变更日志"折叠到每 key 最新值"的物化，流是表"逐次变更"的差异日志——同一条日志的两种视图。

底层机制（比文档深一层）

两个方向都要成立，对偶才成立：

流 → 表（聚合 / 物化）。把一条变更流按 key 折叠，就得到表：相同 key 后值覆盖前值，tombstone 删除该 key。count()、aggregate() 做的就是这件事——它们把一条 KStream"播放"成一张 KTable。

表 → 流（差异 / changelog）。每当表里某个 key 的值变了，就向外发一条"这个 key 现在是这个值"的记录。这串记录就是表的 changelog。KTable.toStream() 做的就是这件事。

关键在于：Streams 不是"另外维护一条日志来记录表的变化"。那条 changelog 就是表的本体。本地 RocksDB 里的表只是这条日志在某一时刻折叠出的物化结果（一份缓存）。崩溃恢复时，新实例从 changelog topic 的头开始回放，逐条折叠，就重建出了崩溃前那张表——不需要快照、不需要 checkpoint。这是 Streams 和 Flink/Spark 在容错模型上的根本分野：那两者靠周期性 checkpoint 落盘整个状态，Streams 靠回放一条始终在 Kafka 里的 changelog。

与下一个概念的关系：流表对偶解释了"状态为什么能恢复"。但"恢复"这个动作发生在谁身上、什么时候触发？答案要先建立 Streams 的运行单位——instance、StreamThread、task、partition，下一节就拆这条链。

1.4应用即库：instance / thread / task / partition

一个 Streams 应用就是一个普通 Java 进程，内部起 N 个 StreamThread，每个线程跑若干 task，每个 task 钉死绑定一组输入分区。

底层机制（比文档深一层）

四级映射，从外到内：

• instance（实例）= 一个跑着你 Streams 代码的 JVM 进程。多开几个进程（多个 pod）就是横向扩容。
• StreamThread（流线程）= 每个实例内起 num.stream.threads 个。每个 StreamThread 内部持有一个普通的 KafkaConsumer，加入消费组，group.id 就是你的 application.id。这是"架在消费组之上"的字面实现。
• task（任务）= 调度的最小单位。task 数 = 各子拓扑输入 topic 的最大分区数。4 个分区的输入 topic，就是 4 个 task，编号 0–3。这个数在拓扑结构定下来的那一刻就钉死了，运行时不可改。
• partition（分区）= 每个 task 绑定一组输入分区（最简单情况下一个 task 对一个分区）。task 和它的分区是焊死的，不会在运行中拆开分给两个线程。

task 在"所有实例 × 所有线程"这个池子里分配。加一个实例，触发消费组再平衡，一部分 task 迁移过去。这里藏着那个最常考的结论：并行上限 = 最大输入分区数。task 总数被分区数钉死，所以能同时干活的线程数不可能超过 task 数；线程开得比 task 多，多出来的线程空转闲置。要再提并行度，唯一办法是给输入 topic 加分区——而那是个破坏性操作（改变 key 的分区落点，§4.3）。

场景走查：4 分区 topic，逐步加线程和实例

• 1 个实例、num.stream.threads=1：1 个线程扛全部 4 个 task。能跑，但没并行。
• 1 个实例、num.stream.threads=4：4 个线程各 1 task，吃满并行。
• 2 个实例、各 2 线程：共 4 线程各 1 task，和上一档算力相同，但抗单实例宕机。
• 再加到共 6 线程：仍只有 4 个 task，2 个线程闲置。要更快只能给输入 topic 加分区。

与下一个概念的关系：task 是按"子拓扑的输入分区"切出来的。"子拓扑"是什么、一个应用怎么被切成几个子拓扑——这要先看清 Streams 把你的 DSL 代码编译成了什么结构，也就是下一节的拓扑。

1.5拓扑：DSL 编译成的处理图

拓扑是 source → processor → sink 组成的有向图；你写的 DSL 被编译成它，子拓扑的边界就是内部 topic。

底层机制（比文档深一层）

拓扑由三类节点构成：source processor（从一个 topic 读入）、stream processor（map/filter/aggregate 等变换）、sink processor（写出到一个 topic）。DSL 的每个算子被编译成一个或多个 processor 节点连起来。

关键的一层深度在子拓扑（sub-topology）边界。一张拓扑会被切成若干子拓扑，切点正是需要重洗数据的地方——也就是改了 key、下游又要按 key 聚合/join 的位置。在切点上，上游子拓扑把数据写到一个内部 topic，下游子拓扑再从这个 topic 读回来。这个内部 topic 就是重分区 topic（§2.3）。所以"子拓扑之间靠内部 topic 连接"和"task 数 = 子拓扑输入分区数"是同一件事的两面：每个子拓扑独立按它的输入分区数切 task。

调用 topology.describe() 能打印出这张图的文本结构，看到它被切成了几个子拓扑、各含哪些 processor。这是确认"运行时到底建了几个内部 topic"的第一手段，比对着 DSL 脑补可靠。

Topology topo = b.build();
System.out.println(topo.describe());  // 打印 source/processor/sink + 子拓扑边界

与下一个概念的关系：图 1.3 里那个把拓扑切成两半的重分区 topic，是被一个"改了 key 的算子 + 下游有状态算子"触发出来的。哪些算子有状态、哪些会改 key——这就是最后一节 stateless vs stateful 的分界。

1.6stateless vs stateful 算子

stateless 算子单条记录就能算出结果；stateful 算子要跨多条记录维护状态，因此需要状态存储，且改 key 时会触发重分区。

底层机制（比文档深一层）

stateless：map、mapValues、filter、flatMap、branch、merge、peek。每条记录的输出只取决于这条记录自己，运行时不为它们分配状态存储，也没有 changelog。

stateful：count、aggregate、reduce、各种 join、所有窗口算子。它们的输出取决于"这个 key 此前累积的状态"，所以必须有一块状态存储记着累积值（§2.1 详解，本节只点到"需要它"）。

两类之间有一条暗线：改 key 的操作会给数据流打上"需重分区"标志。selectKey、map（可能改 key）、flatMap、groupBy 都属此类。一旦改了 key，下游的 stateful 算子要求"同一 key 的所有记录落在同一 task"才能正确聚合——而改 key 后记录的分区落点乱了，于是 Streams 在它们之间自动插一个重分区 topic 把数据按新 key 重洗（§2.3 详解，含为什么 groupByKey 比 groupBy 省、以及 selectKey 为何是 lazy 的）。这就是图 1.3 里那个子拓扑边界的来历。

场景走查：一条链里哪些算子贵

b.stream("clicks")
 .filter((k, v) -> v.valid())          // stateless：零状态，零内部 topic
 .selectKey((k, v) -> v.userId())      // 改 key：打上"需重分区"标志
 .groupByKey()                         // 不再改 key
 .count();                             // stateful：状态存储 + changelog + 触发重分区

承上启下：这六节搭好了 Streams 的词汇表——库的运行单位、两种数据视图、对偶、拓扑、算子分类。每一节都把"机制深一层"压到了一个停止点：状态存储到底怎么落盘、重分区的具体成本、再平衡时状态怎么搬、窗口和时间怎么算、四种 join、EOS 怎么实现——这些是 02 原理 的正题。