如何使用Spring KafkaStreams合并两个Kafka Topic数据并关联生成业务对象

方案合理性结论 #

你提出的方案完全合理，这是Kafka Streams非常典型的适用场景：将两个同主键的数据流通过状态存储暂存，待两侧数据都就绪后再合并，Kafka Streams内置的状态管理、容错机制会帮你处理缓存持久化、节点故障恢复、数据分片等底层逻辑，不需要自己额外实现半消息缓存、超时清理等能力。

实现示例 #

你已经完成了将两个Topic转为状态表（KTable）的基础逻辑，接下来可以通过两种方式实现合并触发逻辑：

方式1：KTable-KTable Inner Join（最简洁，推荐） #

KTable的Join天生支持「任意一侧表数据更新时，若另一侧同主键数据存在，则触发合并输出」的语义，完全匹配你的需求，只需要新增如下逻辑即可：

@Autowired
public void buildOrderMergeStream(StreamsBuilder sb) {
    // 读取已构建的两个状态表
    KTable<Integer, Product> productTable = sb.table("products-view");
    KTable<Integer, Price> priceTable = sb.table("prices-view");

    // 内连接合并，仅当两侧同ID数据都存在时才会输出结果
    productTable.join(
            priceTable,
            // 合并规则：将Product和Price拼接为Order对象
            (product, price) -> {
                Order order = new Order();
                order.setId(product.getId());
                order.setItem(product.getItem());
                order.setPrice(price.getPrice());
                return order;
            }
        )
        // 将合并结果转为流输出到目标Topic，也可直接进行后续处理
        .toStream()
        .to("merged-orders", Produced.with(Serdes.Integer(), Serdes.serdeFrom(Order.class)));
}

方式2：流主动查询状态表（适合需要自定义校验逻辑的场景） #

如果你需要在合并前做更多自定义校验（比如字段合法性校验、业务规则过滤），可以让每个流在收到消息时主动查询另一个状态表：

@Autowired
public void buildProductTriggerStream(StreamsBuilder sb) {
    KStream<Integer, Product> productStream = sb.stream("products", 
        Consumed.with(Serdes.Integer(), Serdes.serdeFrom(Product.class)));

    productStream.process(() -> new Processor<Integer, Product>() {
        private KeyValueStore<Integer, Price> priceStore;

        @Override
        public void init(ProcessorContext context) {
            // 绑定价格状态存储
            priceStore = context.getStateStore("prices-view");
        }

        @Override
        public void process(Integer key, Product product) {
            // 查询对应ID的价格数据是否存在
            Price matchedPrice = priceStore.get(product.getId());
            if (matchedPrice != null) {
                Order mergedOrder = new Order();
                mergedOrder.setId(product.getId());
                mergedOrder.setItem(product.getItem());
                mergedOrder.setPrice(matchedPrice.getPrice());
                // 向下游传递合并结果
                context.forward(mergedOrder.getId(), mergedOrder);
            }
        }

        @Override
        public void close() {}
    }, "prices-view")
    .to("merged-orders", Produced.with(Serdes.Integer(), Serdes.serdeFrom(Order.class)));

    // 同理新增Price流触发逻辑，处理prices topic新消息查询products-view的场景即可
}

注意事项 #

• 需要为自定义的Order类配置序列化/反序列化实现，推荐使用Spring Kafka自带的JsonSerde简化自定义对象的Serde配置
• 可以根据业务允许的半消息最大等待时长，为KTable的状态存储配置保留时间，比如Materialized.as("products-view").withRetention(Duration.ofDays(7))，超时未配对的半消息会自动清理，避免状态存储无限膨胀
• 若业务支持消息更新，KTable Join会自动在任意一侧数据更新时输出最新的合并结果，无需额外处理变更逻辑

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