使用 Spring Cloud Streams 时如何在代码中设置 Kafka Streams 属性？答案

【问题标题】：How to set Kafka Streams properties in code when using Spring Cloud Streams?使用 Spring Cloud Streams 时如何在代码中设置 Kafka Streams 属性？
【发布时间】：2019-06-24 09:19:51
【问题描述】：

在带有 Kafka 的 Spring Boot 中，我可以为 ConsumerFactory 设置属性，如下所示：

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

@Bean
public ConsumerFactory<String, EnrichedOrder> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "barnwaldo");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, EnrichedOrderDeserializer.class);

    return new DefaultKafkaConsumerFactory<>(props);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, EnrichedOrder> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, EnrichedOrder> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
}

}

使用 Kafka Streams，我可以在代码中设置属性如下：

    final Properties streamsConfiguration = new Properties();
    streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-lambda-example");
    streamsConfiguration.put(StreamsConfig.CLIENT_ID_CONFIG, "wordcount-lambda-example-client");
    streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
    streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
    final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);

使用 Spring Cloud Streams 和 Kafka Streams 时，所有属性似乎只能通过资源文件夹中的 application.properties 或 application.yml 文件输入，例如

spring.cloud.stream.bindings:
    output:
        contentType: application/json
        destination: data2
    input:
        contentType: application/json
        destination: data1
spring.cloud.stream.kafka.streams:
    binder:
      brokers: localhost
      configuration:
        commit.interval.ms: 1000
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    bindings.input.consumer:
        applicationId: data-tester

当使用 Spring Cloud Streams 和 Kafka Streams 时，有没有办法在 HashMap 或 Properties 中包含属性。

也许这可以通过 KafkaMessageChannelBinder 或通过扩展 AbstractMessageChannelBinder 以某种方式完成 - 请参阅https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/blob/7355ada4613ad50fe95430f1859d4ea65f004be1/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java。

我找不到与此相关的文档；非常感谢任何帮助。

【问题讨论】：

标签： apache-kafka stream