如何创建带状态存储的多KStream以按Key返回最新值

1. 基础配置初始化 #

首先创建Kafka Streams核心配置，这是流处理的基础：

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "multi-topic-stream-processor");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-brokers:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
// 第一次启动时消费所有历史消息
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// 设置并行处理线程数，建议匹配所有输入主题的总分区数
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3);

2. 遍历主题列表创建并处理每个KStream #

遍历主题列表，为每个主题单独创建KStream，通过groupByKey+reduce实现每个Key仅保留最新值的状态存储，最后统一输出到outputTopic：

List<String> topics = Arrays.asList("Topic1", "Topic2", "Topic3");
StreamsBuilder builder = new StreamsBuilder();

for (String topic : topics) {
    // 从指定主题创建KStream
    KStream<String, String> sourceStream = builder.stream(topic);
    
    // 定义唯一的状态存储名称，避免不同主题的状态冲突
    String storeName = topic + "-latest-value-store";
    
    // 通过reduce操作让新值覆盖旧值，实现每个Key仅保留最新值，同时生成对应状态存储
    KTable<String, String> latestValueTable = sourceStream
            .groupByKey()
            .reduce(
                (oldValue, newValue) -> newValue,
                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as(storeName)
                        .withKeySerde(Serdes.String())
                        .withValueSerde(Serdes.String())
            );
    
    // 将KTable转回KStream，输出到目标主题
    latestValueTable.toStream().to("outputTopic");
}

3. 启动和管理Streams应用 #

初始化并启动Kafka Streams实例，处理异步消费与消息输出：

KafkaStreams streams = new KafkaStreams(builder.build(), props);

// 添加关闭钩子，实现优雅停机
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

// 启动应用
streams.start();

关键注意事项 #

• 状态存储隔离：每个主题的状态存储名称必须唯一，否则会出现不同主题的Key互相覆盖的问题。
• 并行处理优化：NUM_STREAM_THREADS_CONFIG的设置建议匹配所有输入主题的总分区数，最大化并行处理能力，高效消化百万级消息。
• 历史消息消费：第一次启动时务必设置AUTO_OFFSET_RESET_CONFIG为earliest，确保能消费主题中已存在的所有历史消息。
• 状态持久化：默认使用RocksDB作为状态存储，会自动持久化到磁盘，重启后不会丢失状态。

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