《Java 消息中间件·事件驱动架构》8.1 表与流的互转

同一个 topic，两种看法

Kafka Streams 不只有一种「消费者」。对同一条消息序列，你可以把它当成不可追加的流（KStream），也可以当成可查询的表（KTable）。这两个抽象不是 API 层面的命名游戏，而是底层状态管理方式的根本分歧。

在 libraryd（本书贯穿的合成项目）里，一笔借阅从创建到归还会经过多个状态。如果把它当 KStream 消费，每次状态变更都会独立向下游传递；如果把它当 KTable 消费，下游永远只看到该借阅当前的最新状态。

KStream 与 KTable 的算子行为差异

先看一个纯词法上的对比。对同一个输入 ("L-1", "B-101|BORROWED")，两个算子的后续行为完全不同。

// KStream：filter 只过滤本条记录，不维护任何状态
KStream<String, String> stream = builder.stream("loan-events");
stream.filter((k, v) -> v.contains("RETURNED")); // 只匹配当前这条

// KTable：groupByKey + reduce 维护每个 key 的最新值
KTable<String, String> table = stream
    .groupByKey()
    .reduce((oldVal, newVal) -> newVal); // 同 key 覆盖

KStream 的 filter 是无状态的：它不需要记住之前见过什么，也不需要保留任何运行时数据结构。KTable 的 reduce 是有状态的：它必须在本地维护一个键值存储，每条新记录进来都要查旧值、写新值。

这个状态存储在 Kafka Streams 里默认用 RocksDB 实现，撑在本地磁盘上。遇到重启或再平衡时，Streams 会从 changelog topic 恢复状态，不需要重新从头消费原始 topic。但这也意味着：KTable 的算子天然引入状态存储开销，而 KStream 的纯变换算子没有。

toTable() 与 toStream() 的语义代价

从 KStream 到 KTable 的转换，本质是把「无界事件流」物化为「有界快照表」。反过来从 KTable 到 KStream 的转换，是把「快照的每次变更」重新变成一条事件记录。两个方向的转换都有代价。

toTable：从流到表

输入是同 key 的多条记录，输出只有该 key 的最新值。

KTable<String, String> loanTable = loanStream
    .groupByKey()
    .reduce((oldVal, newVal) -> newVal);

代价：

• 引入状态存储，需要磁盘空间和恢复时间；
• 输入的顺序敏感，乱序到达的事件可能覆盖「更晚但先到的」值；
• 如果原始 topic 不是 log-compacted，同 key 的历史记录会被保留两份：一份在原始 topic，一份在 changelog topic。

toStream：从表到流

KStream<String, String> backToStream = loanTable.toStream();

代价：

• 表中的每条输出都带有「变更」语义，但它只有新值，没有旧值（除非用 toStream().mapValues() 手动把旧值也带上）；
• 表的初始恢复阶段可能产生大量「变更」记录，因为状态重建时每个 key 的最快照值都会作为第一条变更被发出去；
• 从 KTable 转回 KStream 后，key 不存在的记录在表阶段已经被过滤（reduce 的初始值如果是 null，不会输出）。

实验：stream -> table -> stream 的完整拓扑

下面用 TopologyTestDriver 真跑一遍，展示两个转换阶段各自的输出。实验环境：Temurin 25 / Kafka 4.3.0 / kafka-streams 4.3.0。

// snippet: StreamTableDualityDemo.java
// 省略 import

public class StreamTableDualityDemo {
    static final String INPUT_TOPIC = "arch08-1.loan-events";
    static final String OUTPUT_TOPIC = "arch08-1.loan-combined";
    static final String STORE_NAME = "loan-snapshot-store";

    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> loanStream = builder.stream(INPUT_TOPIC);
        KTable<String, String> loanTable = loanStream
            .groupByKey()
            .reduce((oldVal, newVal) -> newVal,
                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>
                    as(STORE_NAME));
        KStream<String, String> backToStream = loanTable.toStream();
        backToStream.to(OUTPUT_TOPIC);

        try (TopologyTestDriver driver = new TopologyTestDriver(
                builder.build(), props())) {
            // ...
        }
    }
}

输入事件（libraryd 的借阅记录，key = 借阅单号 loanId）：

  IN  L-1 -> B-101|BORROWED
  IN  L-2 -> B-202|BORROWED
  IN  L-1 -> B-101|RENEWED      (同一笔借阅的状态更新)
  IN  L-3 -> B-303|BORROWED
  IN  L-1 -> B-101|RETURNED     (再次更新)

第一阶段：KStream 输出（每个事件独立成为记录）

  OUT L-1 -> B-101|BORROWED
  OUT L-2 -> B-202|BORROWED
  OUT L-1 -> B-101|RENEWED
  OUT L-3 -> B-303|BORROWED
  OUT L-1 -> B-101|RETURNED

5 条输入产生 5 条输出，每条都独立。KStream 保证「输入一条、输出一条」（除非被 filter/branch 截掉）。L-1 出现了三次，下游收到三次，需要自己决定要不要做状态合并。

第二阶段：KTable 快照（查询状态存储）

  L-1 -> B-101|RETURNED
  L-2 -> B-202|BORROWED
  L-3 -> B-303|BORROWED

状态存储 loan-snapshot-store 里只剩 3 条记录。L-1 的三次更新被压缩成最后一个值 RETURNED。这就是 KTable 的核心语义：同 key 只保留最新快照。

拓扑的自动命名

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [arch08-1.loan-events])
      --> KSTREAM-REDUCE-0000000001
    Processor: KSTREAM-REDUCE-0000000001 (stores: [loan-snapshot-store])
      --> KTABLE-TOSTREAM-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Processor: KTABLE-TOSTREAM-0000000002 (stores: [])
      --> KSTREAM-SINK-0000000003
      <-- KSTREAM-REDUCE-0000000001
    Sink: KSTREAM-SINK-0000000003 (topic: arch08-1.loan-combined)
      <-- KTABLE-TOSTREAM-0000000002

拓扑说明三点：

• KSTREAM-SOURCE 读输入 topic，这是流的起点；
• KSTREAM-REDUCE 持有状态存储，负责把流聚合成表；
• KTABLE-TOSTREAM 把表的变更重新转回流，不持有自己的状态存储。

什么时候用流，什么时候用表

这两个选择不是互斥的，而是同一套数据的不同消费视图。

一个常见的误区是：把事件溯源的 topic 直接当 KTable 消费。如果 topic 没有被 log-compact，同 key 会不断追加，KTable 的 reduce 始终只保留最新一条——你的审计意图就丢了。事件溯源 topic 应该用 KStream 消费，如果需要物化视图，再单独建一个 log-compacted topic 给 KTable。

RocksDB 状态存储的内部结构

KTable 的状态存储用 RocksDB 做底层持久化。RocksDB 是一个基于 LSM 树的嵌入式存储，写操作先追加到 WAL（Write-Ahead Log），再进入内存 memtable，memtable 满了之后按序刷到磁盘（Sorted String Table）。这种结构非常适合 KTable 的工作模式：写多、查少、同 key 覆盖。

KTable update("L-1", "B-101|RENEWED")
  -> WAL 追加 (append)
  -> memtable 插入/覆盖
  -> 同 key 的旧记录在 memtable 被覆盖
  -> WAL + changelog topic 的既有记录保证不丢

RocksDB 的一个关键配置是 state.dir（默认 /tmp/kafka-streams）。每个 Streams 实例的每个 task 在 state.dir 下都有一个独立的子目录，命名格式是 <appId>/<taskId>/<store-name>。这个目录在实例启动时恢复、在关闭时不会自动清理。生产环境里应该把它挂载在独立的 SSD 盘上，避免与系统盘抢 I/O。

changelog topic 和 RocksDB 不是二选一的关系，而是互为备份：

• changelog topic 在 Kafka 集群上，跨节点冗余，用于 task 迁移或实例重启时的远程恢复；
• RocksDB 本地文件 在实例本地，用于亚毫秒级查询；
• 重启时，如果本地 RocksDB 文件完好，优先用本地文件恢复；如果文件损坏或被删，则从 changelog topic 恢复。

这意味着changelog topic 的保留策略必须大于「实例最长停机容忍时间」。如果 retention 是 7 天，但实例停了 10 天，本地文件也丢了，恢复时就会从 topic head 读到 tail，但中间 3 天的数据已经被 retention 截掉，状态恢复就不完整了。

toStream() 的初始输出与更新语义差异

KTable.toStream() 的输出看起来和原始 KStream 一样，都是「key -> value」的记录，但语义完全不同。原始 KStream 的每条记录是「一个独立的事实」；toStream() 的每条记录是「这个 key 的最新值变了」。

这在实际消费中有两个具体区别。

第一，初始恢复阶段的「虚假变更」。 当应用第一次启动、RocksDB 为空时，Streams 会从 changelog topic 的 head 消费所有记录来重建状态。重建完成后，toStream() 会把每个 key 的最终快照值作为第一批「变更事件」发出去。如果 changelog topic 里有 100 万个 key，启动后你会立刻看到 100 万条输出。这不是「100 万个新事实」，而是「100 万个当前值的快照」。

第二，后续增量只有「diff」。 在恢复完成后，只有真正收到新消息时 toStream() 才会输出。如果一条新消息的 value 与上一个快照完全一致（比如同样是 B-101|BORROWED），reduce 的旧值等于新值，toStream() 不会输出任何记录。这与 KStream 的行为形成对比：KStream 即使收到相同的 value 也会原样转发。

// KStream：相同 value 也会输出
loanStream.filter((k, v) -> true); // 5 条输入，5 条输出（value 可重复）

// KTable -> toStream()：相同 value 不触发输出
KTable<String, String> table = loanStream.groupByKey().reduce((o, n) -> n);
table.toStream(); // 仅当新值 != 旧值时才输出

Log compaction 的角色与副作用

KTable 背后的状态存储恢复依赖 changelog topic。Kafka Streams 在向 KTable 写入状态时，会自动创建以 <app-id>-<store-name>-changelog 命名的内部 topic，并启用 log compaction。

Log compaction 的本质是：同 key 的旧记录被标记为删除，只保留最新一条。这与 KTable 的语义天然对齐，但带来两个副作用：

1. 磁盘抖动：compaction 是后台线程周期性执行的，大吞吐量场景下，磁盘 I/O 会出现明显的「写峰-清理谷」交替。
2. 恢复带宽：应用重启后，Streams 必须从 changelog topic 的 head 读到 tail 来重建状态。如果 changelog topic 很大（几十 GB），实例重启的「冷启动」时间会很长。卷一的 10.2 节详细讨论过基于 offset 的快照恢复——生产环境应该启用，否则每次重启都从头读几十 GB 是不现实的。

Tombstone：KTable 中 key 的删除语义

KTable 的 value 如果是 null（即发送一条 key 存在但 value 为空的消息），这条记录在语义上表示删除。RocksDB 会把它清理掉，后续的 toStream() 也会把这条 null 值作为「删除事件」转发给下游。

在 libraryd 场景里，一笔借阅记录如果在归档后需要清理，可以发一条 ("L-1", null) 到原始 topic。KTable 消费后会把 L-1 从快照里移除，toStream() 阶段会输出一条 L-1 -> null 的删除事件。下游的索引系统可以据此把这条记录从 Elasticsearch 等搜索索引里删掉。

// The following null value tells KTable to remove the key
input.pipeInput("L-1", null);

注意：如果下游消费者不理解 tombstone，会把 null value 当成数据缺失而不是删除指令，留下「僵尸数据」。集成搜索索引或缓存时，必须把 tombstone 的语义纳入消费端协议。

小结

• KStream 是「无界增量流」，每条记录独立向下游传递，无状态算子（filter/map）不引入持久化开销。
• KTable 是「有界快照表」，同 key 的后续记录覆盖前一条，有状态算子（reduce/aggregate/join）维护 RocksDB 本地存储。
• toTable() 把流物化为快照，代价是状态存储与恢复；toStream() 把快照变更重新变成流，代价是可能丢失旧值上下文。
• 实验表明 5 条输入在 KStream 阶段输出 5 条，在 KTable 阶段合并为 3 条快照，验证了流表对偶的核心语义。
• 事件溯源 topic 避免直接用 KTable，除非启用 log compaction；实时物化视图优先用 KTable。
• RocksDB 本地存储与 changelog topic 互为备份，恢复时优先用本地文件，缺失时才从远程 topic 重建；changelog retention 必须大于最长停机时间。
• toStream() 在初始恢复阶段会输出大量「快照变更」，这不是真实的新事实，而是状态重建的副作用；下游消费者需要区分增量和全量。
• Tombstone（value 为 null）是 KTable 的删除语义，下游消费者必须理解它，否则会留下无法清理的僵尸数据。
• 调试流表互转拓扑时，TopologyTestDriver 和 print(Printed.toSysOut()) 是最高效的组合：前者验证单元逻辑，后者打印运行期数据。
• 流与表的选择不是二选一，而是对同一 topic 的两种消费视图；产品中经常同时存在一条 KStream 消费链和一条 KTable 消费链。

阅读导航：上一节 7.3 窗口与聚合 ｜ 下一节 8.2 KTable 的 Join