springboot 动态监听kafka主题队列

在Spring Boot中，可以使用Spring Kafka来实现动态监听Kafka主题队列。以下是实现步骤：

配置Kafka消费者工厂

在application.properties或application.yml中配置Kafka消费者工厂：

spring.kafka.consumer.bootstrap-servers=<kafka服务器地址>
spring.kafka.consumer.group-id=<消费者组ID>
spring.kafka.consumer.auto-offset-reset=earliest

创建Kafka消费者

创建Kafka消费者，实现KafkaListener注解，用于监听Kafka主题队列。也可以使用@KafkaListeners注解监听多个主题队列。

@Component
public class KafkaConsumer {
 
    @KafkaListener(topics = "${kafka.topic}")
    public void consume(String message) {
        // 处理消息
    }
}

动态添加Kafka主题队列

使用KafkaAdminClient可以动态添加Kafka主题队列。在需要添加主题队列的地方，通过KafkaAdminClient创建新的主题队列：

@Autowired
private KafkaAdminClient adminClient;

public void addTopic(String topicName) {
    NewTopic newTopic = new NewTopic(topicName, 1, (short) 1);
    adminClient.createTopics(Collections.singleton(newTopic));
}

动态监听Kafka主题队列

使用KafkaListenerEndpointRegistry可以动态添加或删除Kafka监听器。在需要动态监听主题队列的地方，通过KafkaListenerEndpointRegistry创建新的监听器：

@Autowired
private KafkaListenerEndpointRegistry registry;

public void addListener(String topicName) {
    MessageListener<String, String> messageListener = new MessageListener<String, String>() {
        @Override
        public void onMessage(ConsumerRecord<String, String> data) {
            // 处理消息
        }
    };
    ContainerProperties containerProperties = new ContainerProperties(topicName);
    containerProperties.setMessageListener(messageListener);
    KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
    container.start();
    registry.registerListenerContainer(container);
}

动态删除Kafka主题队列

使用KafkaAdminClient可以动态删除Kafka主题队列。在需要删除主题队列的地方，通过KafkaAdminClient删除主题队列：

@Autowired
private KafkaAdminClient adminClient;

public void deleteTopic(String topicName) {
    adminClient.deleteTopics(Collections.singleton(topicName));
}

以上就是使用Spring Kafka实现动态监听Kafka主题队列的步骤