Spring Boot 3.1.2：如何用新版Kafka Streams Binder创建KeyValueStore型GlobalKTable？

针对Spring Boot 3.1.2 + Spring Cloud Stream Kafka Streams Binder 4.0.3版本，无需绕过Binder手动创建KafkaStreams实例，以下是基于官方推荐的函数式编程模型实现GlobalKTable并物化为KeyValueStore的方案：

1. 配置文件（application.yaml） #

先通过配置指定GlobalKTable的主题、存储名称及序列化规则：

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            brokers: your-kafka-broker-address:9092
            configuration:
              application.id: your-app-id
              default.key.serde: com.yourpackage.KeySerde # 替换为你的键序列化器
              default.value.serde: com.yourpackage.ValueSerde # 替换为你的值序列化器
      bindings:
        # GlobalKTable的绑定配置，命名规则为{beanName}-in-0
        globalKTable-in-0:
          destination: ${topics.gkt-topic} # 替换为你的GlobalKTable主题名
          consumer:
            materializedAs: my-gkt-store # 指定物化后的KeyValueStore名称
            use-native-decoding: true

2. 创建GlobalKTable并绑定到Binder上下文 #

通过注入Binder提供的StreamsBuilder来创建GlobalKTable，由Binder自动管理KafkaStreams实例的生命周期：

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.Materialized;
import org.apache.kafka.streams.state.Stores;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class GlobalKTableConfiguration {

    public static final String GKT_STORE_NAME = "my-gkt-store";
    public static final String GKT_TOPIC = "your-gkt-topic";

    @Bean
    public KeyValueStore<Key, Value> globalKeyValueStore(StreamsBuilder streamsBuilder) {
        // 创建GlobalKTable并物化到指定的KeyValueStore
        streamsBuilder.globalTable(
            GKT_TOPIC,
            Materialized.<Key, Value, KeyValueStore<org.apache.kafka.streams.state.Bytes, byte[]>>as(GKT_STORE_NAME)
                .withKeySerde(KeySerde.INSTANCE)
                .withValueSerde(ValueSerde.INSTANCE)
        );

        // 返回存储引用，用于后续查询
        return (KeyValueStore<Key, Value>) Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore(GKT_STORE_NAME),
            KeySerde.INSTANCE,
            ValueSerde.INSTANCE
        ).build();
    }
}

3. 在流处理逻辑中使用GlobalKTable #

如果需要将GlobalKTable与其他流关联处理，可以在函数式Bean中注入GlobalKTable：

import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.function.Function;

@Configuration
public class StreamProcessingConfig {

    @Bean
    public Function<KStream<Key, InputData>, KStream<Key, OutputData>> processStream(
            GlobalKTable<Key, Value> globalKTable) {

        // 示例：将输入流与GlobalKTable关联
        return inputStream -> inputStream
            .join(globalKTable,
                (inputKey, inputData) -> inputKey, // 关联键规则
                (inputData, gktValue) -> new OutputData(inputData, gktValue) // 关联后的处理逻辑
            );
    }
}

对应的函数绑定配置需添加到application.yaml：

spring:
  cloud:
    stream:
      bindings:
        processStream-in-0:
          destination: your-input-topic
        processStream-out-0:
          destination: your-output-topic

4. 查询GlobalKTable对应的KeyValueStore #

使用Spring Cloud Stream提供的InteractiveQueryService来查询物化后的存储：

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

@Service
public class GktQueryService {

    private final InteractiveQueryService interactiveQueryService;

    public GktQueryService(InteractiveQueryService interactiveQueryService) {
        this.interactiveQueryService = interactiveQueryService;
    }

    public Value getGktValue(Key key) {
        ReadOnlyKeyValueStore<Key, Value> store =
            interactiveQueryService.getQueryableStore(GlobalKTableConfiguration.GKT_STORE_NAME, QueryableStoreTypes.keyValueStore());
        return store.get(key);
    }
}

关键说明 #

• 无需手动创建和启动KafkaStreams实例，Binder会自动管理其启动、关闭及容错
• 摒弃已弃用的@Input注解，采用官方推荐的函数式编程模型或StreamsBuilder注入方式
• 通过materializedAs配置指定存储名称，确保GlobalKTable物化到指定的KeyValueStore

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