如何基于Spring Cloud Streams的@StreamListener接收对象创建KTable？

刚接触Kafka和Spring Cloud Streams的时候，确实容易被不同的编程模型搞混，我来一步步帮你把现有的代码改成基于KTable的实现，后续转GlobalKTable也会顺理成章~

首先，明确核心概念 #

你现在用的是Spring Cloud Stream的消息通道模型（MessageChannel + @StreamListener），而KTable是Kafka Streams的组件，属于流处理模型。所以第一步要切换到Kafka Streams的绑定方式，才能使用KTable。

步骤1：确保依赖正确 #

首先要确认你的项目里引入了Kafka Streams的Starter依赖（如果用Maven）：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka-streams</artifactId>
</dependency>

步骤2：修改绑定接口 #

把原来的MessageChannel换成Kafka Streams的KStream（因为我们要从输入Topic读取流数据，再转换成KTable）：

import org.springframework.cloud.stream.annotation.Input;
import org.apache.kafka.streams.kstream.KStream;

public interface MyBindings { // 类名建议首字母大写，符合Java规范
    String INPUT = "input";
    String PROCESSING_TABLE_STORE = "processing-store"; // 后续用于状态存储的名称

    @Input(INPUT)
    KStream<String, String> input(); // 泛型是<Key类型, Value类型>，这里原始消息是JSON字符串，所以Value用String
}

步骤3：重写消息处理逻辑，构建KTable #

原来的@StreamListener要改成接收KStream，然后我们从JSON中提取userId和appId作为复合键，再将流聚合为KTable（KTable会自动维护一个物化的状态存储，用来跟踪数据是否正在处理）：

import com.fasterxml.jackson.databind.ObjectNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.Input;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.common.serialization.Serdes;
import org.springframework.stereotype.Service;

@Service
public class ProcessingTrackerService {

    private final ObjectMapper objectMapper = new ObjectMapper();

    @StreamListener
    public void trackProcessing(@Input(MyBindings.INPUT) KStream<String, String> inputStream) {
        // 1. 从JSON中提取userId和appId，构建复合键
        KStream<String, String> keyedStream = inputStream
                .map((ignoredKey, jsonAsString) -> {
                    try {
                        ObjectNode node = objectMapper.readValue(jsonAsString, ObjectNode.class);
                        String userId = node.get(Info.USER_ID_KEY).asText();
                        String appId = node.get(Info.APP_ID_KEY).asText();
                        // 用"userId:appId"作为复合键，确保唯一标识一个组合
                        String compositeKey = userId + ":" + appId;
                        // 值可以存储你需要的状态，比如"processing"
                        return new KeyValue<>(compositeKey, "processing");
                    } catch (Exception e) {
                        // 处理JSON解析异常，比如跳过无效消息
                        throw new RuntimeException("Failed to parse message JSON", e);
                    }
                });

        // 2. 将KStream转换为KTable，物化到状态存储中
        KTable<String, String> processingTable = keyedStream
                .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
                .reduce(
                        (oldStatus, newStatus) -> newStatus, // 用最新状态覆盖旧状态
                        Materialized.as(MyBindings.PROCESSING_TABLE_STORE) // 指定状态存储名称，后续查询用
                );

        // （可选）如果需要把KTable的状态同步到另一个Topic，可以添加这行
        // processingTable.toStream().to("processing-status-topic");
    }
}

步骤4：配置Kafka Streams属性 #

在application.yml（或application.properties）里添加必要的配置，确保Kafka Streams正常工作：

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            application-id: processing-tracker-app # Kafka Streams应用ID，全局唯一
            brokers: localhost:9092 # 你的Kafka Broker地址
            configuration:
              commit.interval.ms: 1000 # 状态存储的提交间隔，影响查询实时性
      bindings:
        input:
          destination: your-input-topic # 你要监听的输入Topic名称
          consumer:
            value-serde: org.apache.kafka.common.serialization.Serdes$StringSerde

步骤5：查询KTable的状态（实现你的核心需求） #

要查询userId+appId组合是否正在处理，你可以通过**交互式查询（Interactive Query）**来访问KTable的状态存储：

import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.springframework.stereotype.Component;

@Component
public class ProcessingStatusChecker {

    private final InteractiveQueryService interactiveQueryService;

    // 注入InteractiveQueryService
    public ProcessingStatusChecker(InteractiveQueryService interactiveQueryService) {
        this.interactiveQueryService = interactiveQueryService;
    }

    public boolean isProcessing(String userId, String appId) {
        String compositeKey = userId + ":" + appId;
        // 获取状态存储
        ReadOnlyKeyValueStore<String, String> store = interactiveQueryService
                .getQueryableStore(MyBindings.PROCESSING_TABLE_STORE, QueryableStoreTypes.keyValueStore());
        // 检查键是否存在，存在则说明正在处理
        return store.get(compositeKey) != null;
    }
}

后续转GlobalKTable的思路 #

如果后续要切换到GlobalKTable，其实非常简单：

1. 绑定接口的输入可以改为GlobalKTable：

   @Input(INPUT)
   GlobalKTable<String, String> input();
   

2. GlobalKTable会在每个应用实例中加载整个Topic的全量数据，适合跨实例的全局查询场景，查询方式和普通KTable类似，只是状态存储是全局同步的。

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