SpringBoot集管理⼯具KafkaAdminClient使⽤⽅法解析原理介绍
在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.quests包中，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("", 33554432); // 缓存⼤⼩，根据本机内存⼤⼩配置
props.put("linger.ms", 1000); // 发送频率，满⾜任务⼀个条件发送*/
props.put("key.serializer", "org.apache.kafkamon.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafkamon.serialization.StringSerializer");
return props;
}
}
@RestController
public class KafkaTopicManager {
@Autowired
private KafkaConfig kafkaConfig;
@GetMapping("createTopic")
public void createTopic(){
AdminClient adminClient = Props());
NewTopic newTopic = new NewTopic("test1",4, (short) 1);
Collection<NewTopic> newTopicList = new ArrayList<>();
newTopicList.add(newTopic);
adminClient.close();
}
@GetMapping("deleteTopic")
public void deleteTopic(){
AdminClient adminClient = Props());
adminClient.deleteTopics(Arrays.asList("test1"));
adminClient.close();
}
@GetMapping("listAllTopic")
public void listAllTopic(){
AdminClient adminClient = Props());
ListTopicsResult result = adminClient.listTopics();
KafkaFuture<Set<String>> names = result.names();
try {
<().forEach((k)->{
System.out.println(k);
springboot中文
});
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
adminClient.close();
}
@GetMapping("getTopic")
public void getTopic(){
AdminClient adminClient = Props());
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();
}
}
以上就是本⽂的全部内容，希望对⼤家的学习有所帮助，也希望⼤家多多⽀持。

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。