apache-kafka - 使用 Spring Cloud Streams 时如何在代码中设置 Kafka Streams 属性?
问题描述
在带有 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。
我找不到与此相关的文档;任何帮助是极大的赞赏。
解决方案
默认情况下,它支持绑定器级别,其中属性应以spring.cloud.stream.kafka.streams.binder.
文字 为前缀
如果你看到这个KafkaStreamsBinderSupportAutoConfiguration
类,你可以看到 bean 配置,它从 yaml 属性中读取并设置为 kafka 流。
推荐阅读
- android - 如何更改 TextView 的 drawableEnd 属性
- android - 不让我构建我的 android 项目,但让我在 Unity 2019.1.6f1 Personal 中调试它
- python - 是否有图书馆或建议的策略来安排工作时间和休息时间?
- javascript - 为什么 axios.get 没有返回承诺
- angular - Angular 8:上传文件
- r - 提取一系列网格单元
- firebase - 在 Firebase 实时数据库中自定义 ID
- python - 为什么裁剪后的图像颜色会变深?
- scala - 如何让 IntellijIdea 识别我的 build.sc 文件中的语法?
- javascript - 谷歌表格:“如果项目“缺货”,则从数据验证列表中删除项目