SpringBoot如何獲取Kafka的Topic列表

第SpringBoot如何獲取Kafka的Topic列表目錄寫在前面命令行模式代碼模式總結

寫在前面

眾所周知，kafka是現(xiàn)代流行的消息隊列，它使用經(jīng)典的消息訂閱發(fā)布模式實現(xiàn)消息的流轉，大部分代碼結合kafka使用都是使用它的生產(chǎn)者和消費者來實現(xiàn)消息的傳遞，那么對于kafka的主題的管理怎么使用代碼實現(xiàn)呢，這是今天要講的主題

命令行模式

kafka要結合zookeeper使用，因為它把元數(shù)據(jù)信息交給了zookeeper管理，其實使用命令行命令很容易就能對topic進行管理，主要使用的命令是kafka-topics.sh

創(chuàng)建主題

kafka-topics.sh--zookeeperlocalhost:2181--create--topicmy-topic--replication-factor3--partitions3

查看主題列表

kafka-topics.sh--zookeeperlocalhost:2181--list

查看主題狀態(tài)

kafka-topics.sh--describe

--zookeeper:2181--topicTestTopic

代碼模式

那么話說回來如何使用代碼實現(xiàn)topic的管理呢，那么現(xiàn)在就來看一下代碼的實現(xiàn)方式，此處使用springboot2框架實現(xiàn)。

首先引進依賴kafka的相關

dependency

groupIdorg.springframework.kafka/groupId

artifactIdspring-kafka/artifactId

/dependency

創(chuàng)建一個測試類進行測試

publicstaticvoidmain(String[]args){

Propertiesproperties=

newProperties();

properties.put("bootstrap.servers","1:9093");

properties.put("key.serializer","mon.serialization.StringSerializer");

properties.put("value.serializer","mon.serialization.StringSerializer");

AdminClientadminClient=AdminClient.create(properties);

ListTopicsResultresult=adminClient.listTopics();

KafkaFutureSetStringnames=s();

try{

names.get().forEach((k)-{

System.out.println(k);

});

}catch(InterruptedException|ExecutionExceptione){

e.printStackTrace();

}

adminClient.close();

}

這里面最主要的就是AdminClient這個類，AdminClient實現(xiàn)了Admin接口，Admin里面定義了許多和kafka配置相關的東西

讓我們依次來看一下

publicabstractclassAdminClientimplementsAdmin{

publicAdminClient(){

}

publicstaticAdminClientcreate(Propertiesprops){

return(AdminClient)Admin.create(props);

}

publicstaticAdminClientcreate(MapString,Objectconf){

return(AdminClient)Admin.create(conf);

}

}

而Admin接口里有以下方法

staticAdmincreate(Propertiesprops){

returnKafkaAdminClient.createInternal(newAdminClientConfig(props,true),(TimeoutProcessorFactory)null);

}

staticAdmincreate(MapString,Objectconf){

returnKafkaAdminClient.createInternal(newAdminClientConfig(conf,true),(TimeoutProcessorFactory)null);

}

defaultvoidclose(){

this.close(9223372036854775807L,TimeUnit.MILLISECONDS);

}

/**@deprecated*/

@Deprecated

defaultvoidclose(longduration,TimeUnitunit){

this.close(Duration.ofMillis(unit.toMillis(duration)));

}

voidclose(Durationvar1);

defaultCreateTopicsResultcreateTopics(CollectionNewTopicnewTopics){

returnthis.createTopics(newTopics,newCreateTopicsOptions());

}

CreateTopicsResultcreateTopics(CollectionNewTopicvar1,CreateTopicsOptionsvar2);

defaultDeleteTopicsResultdeleteTopics(CollectionStringtopics){

returnthis.deleteTopics(topics,newDeleteTopicsOptions());

}

DeleteTopicsResultdeleteTopics(CollectionStringvar1,DeleteTopicsOptionsvar2);

defaultListTopicsResultlistTopics(){

returnthis.listTopics(newListTopicsOptions());

}

ListTopicsResultlistTopics(ListTopicsOptionsvar1);

defaultDescribeTopicsResultdescribeTopics(CollectionStringtopicNames){

returnthis.describeTopics(topicNames,newDescribeTopicsOptions());

}

DescribeTopicsResultdescribeTopics(CollectionStringvar1,DescribeTopicsOptionsvar2);

defaultDescribeClusterResultdescribeCluster(){

returnthis.describeCluster(newDescribeClusterOptions());

}

DescribeClusterResultdescribeCluster(DescribeClusterOptionsvar1);

defaultDescribeAclsResultdescribeAcls(AclBindingFilterfilter){

returnthis.describeAcls(filter,newDescribeAclsOptions());

}

DescribeAclsResultdescribeAcls(AclBindingFiltervar1,DescribeAclsOptionsvar2);

通過名稱我們可以看出，里面有創(chuàng)建Topic，有刪除Topic，有列出所有Topic,有描述Topic

我們通過這些方法可