SpringBoot如何获取Kafka的Topic列表

2024-04-02 19:55

短信预约 -IT技能 免费直播动态提醒

写在前面

众所周知，kafka是现代流行的消息队列，它使用经典的消息订阅发布模式实现消息的流转，大部分代码结合kafka使用都是使用它的生产者和消费者来实现消息的传递，那么对于kafka的主题的管理怎么使用代码实现呢，这是今天要讲的主题

命令行模式

kafka要结合zookeeper使用，因为它把元数据信息交给了zookeeper管理，其实使用命令行命令很容易就能对topic进行管理，主要使用的命令是kafka-topics.sh

创建主题
kafka-topics.sh --zookeeper localhost:2181 --create --topic my-topic --replication-factor 3 --partitions 3
查看主题列表
kafka-topics.sh --zookeeper localhost:2181 --list
查看主题状态
kafka-topics.sh --describe  --zookeeper 127.0.0.1:2181 --topic TestTopic

代码模式

那么话说回来如何使用代码实现topic的管理呢，那么现在就来看一下代码的实现方式，此处使用springboot2框架实现。

首先引进依赖kafka的相关

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

创建一个测试类进行测试

public static void main(String[] args) {
        Properties properties =  new Properties();
        properties.put("bootstrap.servers", "10.0.59.11:9093");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        AdminClient adminClient = AdminClient.create(properties);
        ListTopicsResult result = adminClient.listTopics();
        KafkaFuture<Set<String>> names = result.names();
        try {
            names.get().forEach((k)->{
                System.out.println(k);
            });
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        adminClient.close();
    }

这里面最主要的就是AdminClient这个类，AdminClient实现了Admin接口，Admin里面定义了许多和kafka配置相关的东西

让我们依次来看一下

public abstract class AdminClient implements Admin {
    public AdminClient() {
    }

    public static AdminClient create(Properties props) {
        return (AdminClient)Admin.create(props);
    }

    public static AdminClient create(Map<String, Object> conf) {
        return (AdminClient)Admin.create(conf);
    }
}

而Admin接口里有以下方法

static Admin create(Properties props) {
        return KafkaAdminClient.createInternal(new AdminClientConfig(props, true), (TimeoutProcessorFactory)null);
    }

    static Admin create(Map<String, Object> conf) {
        return KafkaAdminClient.createInternal(new AdminClientConfig(conf, true), (TimeoutProcessorFactory)null);
    }

    default void close() {
        this.close(9223372036854775807L, TimeUnit.MILLISECONDS);
    }

    
    @Deprecated
    default void close(long duration, TimeUnit unit) {
        this.close(Duration.ofMillis(unit.toMillis(duration)));
    }

    void close(Duration var1);

    default CreateTopicsResult createTopics(Collection<NewTopic> newTopics) {
        return this.createTopics(newTopics, new CreateTopicsOptions());
    }

    CreateTopicsResult createTopics(Collection<NewTopic> var1, CreateTopicsOptions var2);

    default DeleteTopicsResult deleteTopics(Collection<String> topics) {
        return this.deleteTopics(topics, new DeleteTopicsOptions());
    }

    DeleteTopicsResult deleteTopics(Collection<String> var1, DeleteTopicsOptions var2);

    default ListTopicsResult listTopics() {
        return this.listTopics(new ListTopicsOptions());
    }

    ListTopicsResult listTopics(ListTopicsOptions var1);

    default DescribeTopicsResult describeTopics(Collection<String> topicNames) {
        return this.describeTopics(topicNames, new DescribeTopicsOptions());
    }

    DescribeTopicsResult describeTopics(Collection<String> var1, DescribeTopicsOptions var2);

    default DescribeClusterResult describeCluster() {
        return this.describeCluster(new DescribeClusterOptions());
    }

    DescribeClusterResult describeCluster(DescribeClusterOptions var1);

    default DescribeAclsResult describeAcls(AclBindingFilter filter) {
        return this.describeAcls(filter, new DescribeAclsOptions());
    }

    DescribeAclsResult describeAcls(AclBindingFilter var1, DescribeAclsOptions var2);

通过名称我们可以看出，里面有创建Topic，有删除Topic，有列出所有Topic,有描述Topic

我们通过这些方法可以管理Kafka的Topic

最后我们来看一下实现效果