SpringBoot如何获取Kafka的Topic列表
作者:AnthonyJing 时间:2023-11-26 16:01:52
写在前面
众所周知,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个说明代码没问题
来源:https://blog.csdn.net/qq_25464557/article/details/108290280
标签:SpringBoot,Kafka,Topic
![](/images/zang.png)
![](/images/jiucuo.png)
猜你喜欢
Android:如何编写“万能”的Activity
2023-08-15 17:49:26
Java 反射调用静态方法的简单实例
2021-10-09 06:47:58
Java创建线程池为什么一定要用ThreadPoolExecutor
2023-04-22 06:03:31
![](https://img.aspxhome.com/file/2023/1/64391_0s.webp)
大前端代码重构之事件拦截iOS Flutter Vue示例分析
2021-12-11 20:16:53
![](https://img.aspxhome.com/file/2023/0/133130_0s.jpg)
Java Map接口概述和常用方法详解
2022-07-29 14:19:06
![](https://img.aspxhome.com/file/2023/8/103998_0s.png)
java实现单链表、双向链表
2023-02-09 03:15:59
springboot无法从静态上下文中引用非静态变量的解决方法
2022-03-05 16:50:05
C#中重载重写和覆盖的定义与区别
2022-04-23 09:25:34
Springboot JPA 枚举Enum类型存入到数据库的操作
2023-11-25 16:11:40
![](https://img.aspxhome.com/file/2023/4/60064_0s.jpg)
Java最全文件操作实例汇总
2023-11-14 13:00:17
Java之Rsync并发迁移数据并校验详解
2023-07-17 23:01:22
Unity OnGUI实时显示游戏FPS
2021-10-05 22:53:39
SpringBoot配置shiro安全框架的实现
2023-09-07 19:55:38
vscode编写latex的方法
2022-05-21 13:30:52
![](https://img.aspxhome.com/file/2023/6/108736_0s.jpg)
Android实现关机与重启的几种方式(推荐)
2022-08-31 14:29:54
Java毕业设计实战之教室预订管理系统的实现
2023-03-03 20:38:11
![](https://img.aspxhome.com/file/2023/3/74873_0s.jpg)
java底层JDK Logging日志模块处理细节深入分析
2023-02-04 12:47:31
Java实现高校教务系统
2022-05-16 04:24:17
关于Lambda表达式的方法引用和构造器引用简的单示例
2022-09-18 05:02:08
![](https://img.aspxhome.com/file/2023/1/75681_0s.jpg)
Spring常用注解汇总
2022-12-19 16:26:03