Apache Kafka(v3)：KTable表现类似KStream？技术咨询

为什么会输出所有匹配记录？ #

KTable的核心是维护基于主题数据的状态存储，但在首次启动（或调用cleanUp()后），它会从头消费主题的所有历史数据来构建这个状态表。此时myTable.toStream()输出的是KTable的变更日志流——也就是每个键的每一次更新都会生成一条输出记录，所以你会看到http://www.simpsons.com的两次更新都被打印出来。

只有当状态表完全构建完成后，后续的新消息才会只输出该键的最新值，但初始化阶段的历史更新都会被逐一处理并输出。

如何获取键的最新记录？ #

要直接获取某个键的最终最新值，应该直接查询KTable背后的状态存储，而不是监听变更日志流。具体步骤如下：

1. 修改KTable构建代码，指定可查询的存储名称 #

在创建KTable时，通过Materialized参数指定状态存储的名称，方便后续查询：

KTable<String, String> myTable = builder.table(
    topicName,
    Consumed.with(AutoOffsetReset.EARLIEST),
    Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("simpsons-table-store") // 指定存储名称
);

2. 启动流后，查询状态存储获取最新值 #

等Kafka Streams进入运行状态后，通过存储名称获取只读存储实例，直接查询目标键：

KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), new StreamsConfig(props));

kafkaStreams.cleanUp();
kafkaStreams.start();

// 等待流完全启动并完成状态初始化
while (!kafkaStreams.state().equals(KafkaStreams.State.RUNNING)) {
    Thread.sleep(100);
}

// 获取状态存储
ReadOnlyKeyValueStore<String, String> keyValueStore = kafkaStreams.store(
    StoreQueryParameters.fromNameAndType(
        "simpsons-table-store",
        QueryableStoreTypes.keyValueStore()
    )
);

// 查询目标键的最新值
String latestValue = keyValueStore.get("http://www.simpsons.com");
System.out.println("[KTable Latest]: http://www.simpsons.com, " + latestValue);

Thread.sleep(5000);
kafkaStreams.close();

运行结果 #

此时会只输出：

[KTable Latest]: http://www.simpsons.com, two

补充说明 #

• KTable的toStream()方法本质是输出状态的变更历史，适合监听数据的实时更新场景；
• 如果你的场景是需要定期或按需获取某个键的最新状态，直接查询状态存储是最直接高效的方式；
• 不要在生产环境中频繁调用cleanUp()，它会清空本地状态存储，导致每次启动都要重新从头构建状态，影响性能。

内容的提问来源于stack exchange，提问作者Harinder