Kafka 消费者动态获取主题

Question

我在 Spring Boot 中配置了一个 Kafka 消费者。这是配置类：

@EnableKafka
@Configuration
@PropertySource({"classpath:kafka.properties"})
public class KafkaConsumerConfig {

    @Autowired
    private Environment env;

    @Bean
    public ConsumerFactory<String, GenericData.Record> consumerFactory() {

        dataRiverProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("bootstrap.servers"));
        dataRiverProps.put(ConsumerConfig.GROUP_ID_CONFIG, env.getProperty("group.id"));
        dataRiverProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, env.getProperty("enable.auto.commit"));
        dataRiverProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, env.getProperty("auto.commit.interval.ms"));
        dataRiverProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, env.getProperty("session.timeout.ms"));
        dataRiverProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, env.getProperty("auto.offset.reset"));

        dataRiverProps.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, env.getProperty("schema.registry.url"));
        dataRiverProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
        dataRiverProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());

        return new DefaultKafkaConsumerFactory<>(dataRiverProps);
    }

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

这是消费者：

@Component
public class KafkaConsumer {

    @Autowired
    private MessageProcessor messageProcessor;

    @KafkaListener(topics = "#{'${kafka.topics}'.split(',')}", containerFactory = "kafkaListenerContainerFactory")
    public void consumeAvro(GenericData.Record message) {
        messageProcessor.process();
    }

}

请注意，我使用 topics = "#{'${kafka.topics}'.split(',')}" 从属性文件中获取主题。 这就是我的 kafka.properties 文件的样子：

kafka.topics=pwdChange,pwdCreation
bootstrap.servers=aaa.bbb.com:37900
group.id=pwdManagement
enable.auto.commit=true
auto.commit.interval.ms=1000
session.timeout.ms=30000
schema.registry.url=http://aaa.bbb.com:37800

现在如果我要向订阅中添加一个新主题，比如 pwdExpire，然后修改 prop 文件如下：

kafka.topics=pwdChange,pwdCreation,pwdExpire

有没有办法让我的消费者在不重新启动服务器的情况下开始订阅这个新主题？ 我找到了这篇帖子 Spring Kafka - Subscribe new topics during runtime，但文档中有关于 metadata.max.age.ms 的说明：

我们强制刷新的时间段（以毫秒为单位） 元数据，即使我们没有看到任何分区领导更改 主动发现任何新的代理或分区。

在我看来这行不通。感谢您的帮助！

Answer 1

没有；唯一的方法是使用主题模式；随着新主题的添加（与模式匹配），默认情况下，代理将在 5 分钟后将它们添加到订阅中。

但是，您可以在运行时为新主题添加新的侦听器容器。

另一种选择是将@KafkaListener bean 加载到子应用程序上下文中，并在每次主题更改时重新创建上下文。

编辑

查看KafkaConsumer.subscribe(Pattern pattern)的javadocs...

/**
 * Subscribe to all topics matching specified pattern to get dynamically assigned partitions.
 * The pattern matching will be done periodically against topics existing at the time of check.
 * <p>
 ...