SpringBoot如何获取Kafka的Topic列表
作者:AnthonyJing 发布时间:2023-11-26 16:01:52
标签:SpringBoot,Kafka,Topic
写在前面
众所周知,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
0
投稿
猜你喜欢
- 在 Java 中,所有的异常都有一个共同的祖先 Throwable(可抛出)。Throwable 指定代码中可用异常传播机制通过 Java
- Jmeter 执行Java 请求时,运行结束后报错,Tidying up remote @ Mon Feb 24 19:42:34 CST
- 推荐阅读idea官网下载链接(对应版本号下载):https://www.jetbrains.com/idea/download/other.
- 百度百科说法:Servlet(Server Applet)是Java Servlet的简称,称为小服务程序或服务连接器,用Java编写的服务
- 本文实例为大家分享了JAVASE系统实现抽卡功能的具体代码,供大家参考,具体内容如下先看下文件结构使用到的知识点:看下Client类的实现:
- 一直对invoke和begininvoke的使用和概念比较混乱,这两天看了些资料,对这两个的用法和原理有了些新的认识和理解。 首先
- 介绍MVC(Model-View-Controller)是一种软件架构模式,其中应用程序被划分为三个部分:模型(Model)、视图(View
- 最近在学习springmvc,今天把springmvc 参数绑定给整理一下,也算个学习笔记吧!@RequestParam 绑定单个请求Req
- 本章将具体讲解下拉菜单的交互。使用下拉菜单(Dropdown)插件,您可以向任何组件(比如导航栏、标签页、胶囊式导航菜单、按钮等)添加下拉菜
- Flink中设计了用户自定义函数体系(User Defined Function,UDF),开发人员实现业务逻辑就是开发UDF。一、环境对象
- 最新开发新项目的时候,要做分享项目,要求分享有微信,微信朋友圈,QQ,QQ空间,新浪微博这五个,所分享内容包括,分享纯图片,纯文字,图文类型
- Android 无障碍的全局悬浮窗可以在屏幕上添加 UI 供用户进行快捷操作,可以展示在所有应用程序之上长期展示。另一方面,在一些自动化场景
- 继承什么是继承呢?继承(Inheritance)是一种联结类与类的层次模型。指的是一个类(称为子类、子接口)继承另外的一个类(称为父类、父接
- InputStreamReader 类1、概述转换流 java.io.InputStreamReader ,是Reader的子类,是从字节流
- 本文实例讲述了java实现List中对象排序的方法。分享给大家供大家参考,具体如下:package com.test; import jav
- 1、何为依赖冲突Maven是个很好用的依赖管理工具,但是再好的东西也不是完美的。Maven的依赖机制会导致Jar包的冲突。举个例子,现在你的
- 开发项目的时候,表很多,是不可能一点点的自己去写xml ,dao文件的,这里就需要用到代码的自动生成工具了。第一步:导入jar包,当然,这之
- monitor概念管程,监视器。在操作系统中,存在着semaphore和mutex,即信号量和互斥量,使用基本的mutex进行开发时,需要小
- Java 用反射设置对象的属性值实例详解/** * 用反射设置对象的属性值 * @param obj 需要設置值的對象 * @param f
- foreach嵌套使用if标签对象取值问题最近做项目过程中,涉及到需要在 Mybatis 中 使用 foreach 进行循环读取传入的查询条