我有“n”个kafka主题,其中我需要根据业务逻辑使用来自不同主题的消息。我需要处理/过滤它们并将其发送到service bus。我不想创建n个kafka配置来使用消息。我只想将其作为一个库,在那里我可以简单地外部化属性以选择和配置使用者,这样我就可以将业务逻辑放在我的应用程序中。以前有人做过这种实现吗?请告诉我这种方法的最佳实践。
编辑:这是我的Kafka消费者配置看起来像:
kafkaconsumerconfig.java文件
@Configuration
public class KafkaConsumerConfig {
@Value("${spring.kafka.consumer.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.consumer.key-deserializer}")
private String keyDeserializer;
@Value("${spring.kafka.consumer.value-deserializer}")
private String valueDeserializer;
@Value("${spring.kafka.consumer.group-id}")
private String groupIdConfig;
@Value("${spring.kafka.consumer.auto-offset-reset}")
private String autoOffsetResetConfig;
@Value("${spring.kafka.consumer.client-id}")
private String clientIdConfig;
@Bean
public KafkaReceiver receiver() {
return new KafkaReceiver();
}
@Bean
public ConsumerFactory<?, ?> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> properties = new HashMap<>();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetResetConfig);
return properties;
}
}
1条答案
按热度按时间zf9nrax11#
您可以通过一个消费者订阅任意数量的主题