Spring Kafka:在运行时订阅新的主题模式 [英] Spring Kafka : Subscribe to a new Topic Pattern during Runtime
问题描述
我正在使用@KafkaListener注释在我的应用程序中使用主题.我需要在运行中的使用者中在运行时更改主题模式,以便可以使用与新模式匹配的新主题.
I'm using the annotation @KafkaListener to consume topics in my application. I need to change the topic pattern at runtime in an already running consumer so that new topics that match the new pattern can be consumed.
我尝试了下面的代码,但是它仍然消耗与旧主题模式匹配的主题.在这里,我在应用程序启动时设置了旧主题模式".然后,我使用Spring @Scheduler每10秒将模式更新为"new-topic-pattern".
I tried the below code, but it still consumes the topics matching the old topic pattern. Here, I have set the "old-topic-pattern" at application start-up. Then, I'm updating the pattern to "new-topic-pattern" every 10 seconds using a Spring @Scheduler.
Class "KafkaTopicPatternConfig.java":
@Configuration
public class KafkaTopicPatternConfig {
@Bean
public String kafkaTopicPattern(Environment env) {
logger.info("Getting kafka topic pattern");
String kafkaTopicPattern = "old-topic-pattern";
return kafkaTopicPattern;
}
}
Class "Consumer.java":
@Component
public class Consumer implements ConsumerSeekAware{
@Autowired
@Qualifier("kafkaTopicPattern")
private String kafkaTopicPattern;
@KafkaListener(topicPattern = "#{kafkaTopicPattern}", id = "s4federatorConsumer")
public void processMessage(@Payload ConsumerRecord<String, Object> record,
@Header(KafkaHeaders.OFFSET) Long offset,
@Header(KafkaHeaders.CONSUMER) KafkaConsumer<String, String> consumer,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partitionId) {
//do something with the consumed message
}
@Scheduled(fixedDelay = 10000, initialDelay = 15000)
public void refreshKafkaTopics() {
logger.info("Inside scheduler to refresh kafka topics");
this.kafkaTopicPattern = "new-topic-pattern";
this.kafkaListenerEndpointRegistry.getListenerContainer("s4federatorConsumer").stop();
this.kafkaListenerEndpointRegistry.getListenerContainer("s4federatorConsumer").start();
}
}
推荐答案
您将kafkaTopicPattern设置为-
You are getting kafkaTopicPattern as -
@Qualifier("kafkaTopicPattern")
private String kafkaTopicPattern;
我看到您正在更新--
this.kafkaTopicPattern = "new-topic-pattern";
但是,如果这两个实例位于不同的实例对象中,则不会刷新注入到侦听器中的"kafkaTopicPattern"的原始值.因此,您必须确保使用新的模式刷新侦听器对象.
But the original value for "kafkaTopicPattern" which is injected in listener wont be refreshed by this if these 2 are in different instance objects. So you will have to make sure that the listener objects are refreshed with the new pattern.
这篇关于Spring Kafka:在运行时订阅新的主题模式的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!