Spring Boot集群管理工具KafkaAdminClient使用方法解析
2020-03-24 07:07
1316 查看
原理介绍
在Kafka官网中这么描述AdminClient:The AdminClient API supports managing and inspecting topics, brokers, acls, and other Kafka objects. 具体的KafkaAdminClient包含了一下几种功能(以Kafka1.0.0版本为准):
- 创建Topic:createTopics(Collection<NewTopic> newTopics)
- 删除Topic:deleteTopics(Collection<String> topics)
- 罗列所有Topic:listTopics()
- 查询Topic:describeTopics(Collection<String> topicNames)
- 查询集群信息:describeCluster()
- 查询ACL信息:describeAcls(AclBindingFilter filter)
- 创建ACL信息:createAcls(Collection<AclBinding> acls)
- 删除ACL信息:deleteAcls(Collection<AclBindingFilter> filters)
- 查询配置信息:describeConfigs(Collection<ConfigResource> resources)
- 修改配置信息:alterConfigs(Map<ConfigResource, Config> configs)
- 修改副本的日志目录:alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment)
- 查询节点的日志目录信息:describeLogDirs(Collection<Integer> brokers)
- 查询副本的日志目录信息:describeReplicaLogDirs(Collection<TopicPartitionReplica> replicas)
- 增加分区:createPartitions(Map<String, NewPartitions> newPartitions)
其内部原理是使用Kafka自定义的一套二进制协议来实现,详细可以参见Kafka协议。主要实现步骤:
客户端根据方法的调用创建相应的协议请求,比如创建Topic的createTopics方法,其内部就是发送CreateTopicRequest请求。
客户端发送请求至Kafka Broker。
Kafka Broker处理相应的请求并回执,比如与CreateTopicRequest对应的是CreateTopicResponse。
客户端接收相应的回执并进行解析处理。
和协议有关的请求和回执的类基本都在org.apache.kafka.common.requests包中,AbstractRequest和AbstractResponse是这些请求和回执类的两个基本父类。
代码如下
@Component public class KafkaConfig{ // 配置Kafka public Properties getProps(){ Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); /* props.put("retries", 2); // 重试次数 props.put("batch.size", 16384); // 批量发送大小 props.put("buffer.memory", 33554432); // 缓存大小,根据本机内存大小配置 props.put("linger.ms", 1000); // 发送频率,满足任务一个条件发送*/ props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); return props; } }
@RestController public class KafkaTopicManager { @Autowired private KafkaConfig kafkaConfig; @GetMapping("createTopic") public void createTopic(){ AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps()); NewTopic newTopic = new NewTopic("test1",4, (short) 1); Collection<NewTopic> newTopicList = new ArrayList<>(); newTopicList.add(newTopic); adminClient.createTopics(newTopicList); adminClient.close(); } @GetMapping("deleteTopic") public void deleteTopic(){ AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps()); adminClient.deleteTopics(Arrays.asList("test1")); adminClient.close(); } @GetMapping("listAllTopic") public void listAllTopic(){ AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps()); 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(); } @GetMapping("getTopic") public void getTopic(){ AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps()); DescribeTopicsResult describeTopics = adminClient.describeTopics(Arrays.asList("syn-test")); Collection<KafkaFuture<TopicDescription>> values = describeTopics.values().values(); if(values.isEmpty()){ System.out.println("找不到描述信息"); }else{ for (KafkaFuture<TopicDescription> value : values) { System.out.println(value); } } adminClient.close(); } }
以上就是本文的全部内容,希望对大家的学习有所帮助
您可能感兴趣的文章:
相关文章推荐
- Kafka 0.11客户端集群管理工具AdminClient
- Kafka读取__consumer_offsets和Kafka 0.11客户端管理工具AdminClient
- 15分钟教你使用 Spring Boot Thymeleaf自定义一个的图床管理工具!
- springboot应用管理工具Spring Boot Admin Reference Guide
- Kafka集群管理工具kafka-manager的安装使用
- springboot添加fastjson第三方json解析工具方法(默认jackjson)
- 使用spring-boot-admin对spring-boot服务进行监控的实现方法
- Spring Boot使用FastJson解析JSON数据的方法
- Spring Boot Admin管理监控数据的方法
- Kafka集群管理工具kafka-manager安装使用
- SpringBoot中Json解析方案与自定义配置方法(二)——使用Gson
- spring-boot使用Admin监控应用的方法
- SpringBoot中Json解析方案与自定义配置方法(一)——使用Jackson
- spring boot整合kafka过程解析
- 解决【必须使用“角色管理工具”安装或配置Microsoft .NET Framework 3.5 SP1】的方法
- Spring Boot Admin的使用
- Spring AOP进行日志记录,管理 (使用Spring的拦截器功能获取对action中每个方法的调用情况,在方法调用前和调用后记录相关日志。)
- 数据库管理工具——SQuirreL SQL Client使用入门
- 【Spring实战】----源码解析SessionFactory及Session的管理及getCurrentSession的使用
- (3)spring boot如何使用第三方json解析框架