SpringBoot中如何通过Kafka客户端获取所有Topic的详细列表？

本文共计858个文字，预计阅读时间需要4分钟。

目录- 写在前- 命令行模式- 代码模式- 总结- 写在前- 众人所知，Kafka是现代流式消息队列，它使用经典的发布/订阅模式实现消息的流转。大部分代码模式都整合了Kafka的使用，都是利用其生产者功能。

目录

• 写在前面
• 命令行模式
• 代码模式
• 总结

写在前面

众所周知，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 */     @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

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

控制台打印里面有3个Topic

去服务器命令行验证一下

也是3个说明代码没问题

总结

网上大多数关于kafka的代码实现都是关于生产者和消费者的实现，今天主要是使用一下kakfa的配置管理类，实现对topic的管理，以此记录作为以后工作中的参考。希望能给大家一个参考，也希望大家多多支持自由互联。