KafkaClientAdmin实现原理：
其内部原理是使用Kafka自定义的一套二进制协议来实现，详细可以参见Kafka协议。主要实现步骤：
?? ??? ?客户端根据方法的调用创建相应的协议请求，比如创建Topic的createTopics方法，其内部就是发送CreateTopicRequest请求。
?? ??? ?客户端发送请求至Kafka Broker。
?? ??? ?Kafka Broker处理相应的请求并回执，比如与CreateTopicRequest对应的是CreateTopicResponse。
?? ??? ?客户端接收相应的回执并进行解析处理。
?? ??? ?和协议有关的请求和回执的类基本都在org.apache.kafka.common.requests包中，AbstractRequest和AbstractResponse是这些请求和回执类的两个基本父类
第一步，创建配置类
package com.liu.config;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Properties;
@Configuration
public class KafkaConf {
? ? @Bean
? ? @ConfigurationProperties(prefix = "spring.kafka")
? ? public KafkaServers kafkaServers(){
? ? ? ? return new KafkaServers();
? ? }
? ? @Bean
? ? public KafkaAdminClient kafkaAdminClient(){
? ? ? ? Properties props = new Properties();
? ? ? ? props.put("bootstrap.servers", String.join(",", kafkaServers().getBootStrapServers()));
? ? ? ? return (KafkaAdminClient) KafkaAdminClient.create(props);
? ? }
}配置类中用到了KafkaServers为自定义的读取配置文件中换行后 以 横杠"-"开头的kafka集群数组配置。
?
package com.liu.config;
public class KafkaServers {
? ? private String[] bootStrapServers;
? ? public String[] getBootStrapServers() {
? ? ? ? return bootStrapServers;
? ? }
? ? public void setBootStrapServers(String[] bootStrapServers) {
? ? ? ? this.bootStrapServers = bootStrapServers;
? ? }
}第二步，定义配置引入配置bean
@Autowired
private KafkaAdminClient kafkaAdminClient;
第三步，直接使用：
NewTopic newTopic = new NewTopic(topicName, partitionNum, (short) replicaNum);
kafkaAdminClient.createTopics(Collections.singleton(newTopic));
kafkaAdminClient中其他的方法：
创建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)