kafka动态创建topic怎么实现

在Kafka中，可以使用Kafka AdminClient API来动态创建Topic

首先，确保你已经将Kafka AdminClient依赖添加到项目中。如果你使用的是Maven，可以在pom.xml文件中添加以下依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

接下来，编写一个使用Kafka AdminClient创建Topic的方法：

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaTopicCreator {

    private final AdminClient adminClient;

    public KafkaTopicCreator(String bootstrapServers) {
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        this.adminClient = AdminClient.create(properties);
    }

    public void createTopic(String topicName, int numPartitions, short replicationFactor) throws ExecutionException, InterruptedException {
        NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
        CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(newTopic));
        createTopicsResult.all().get();
    }

    public void close() {
        adminClient.close();
    }

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        KafkaTopicCreator topicCreator = new KafkaTopicCreator(bootstrapServers);

        try {
            topicCreator.createTopic("my-new-topic", 3, (short) 1);
            System.out.println("Topic created successfully");
        } catch (ExecutionException | InterruptedException e) {
            System.err.println("Failed to create topic: " + e.getMessage());
        } finally {
            topicCreator.close();
        }
    }
}

在这个示例中，我们创建了一个名为KafkaTopicCreator的类，它接受一个bootstrapServers参数，用于连接到Kafka集群。createTopic方法接受三个参数：topicName（Topic名称）、numPartitions（分区数量）和replicationFactor（副本因子）。

在main方法中，我们实例化了一个KafkaTopicCreator对象，并调用createTopic方法来创建一个新的Topic。最后，我们关闭了AdminClient。

请注意，这个示例仅用于演示目的。在实际应用中，你可能需要根据实际需求对代码进行调整，例如添加异常处理、配置参数验证等。