Spring Boot集群管理工具KafkaAdminClient

栏目: IT技术 · 发布时间: 4年前

内容简介:能与原理介绍在Kafka官网中这么描述AdminClient:The AdminClient API supports managing and inspecting topics, brokers, acls, and other Kafka objects. 具体的KafkaAdminClient包含了一下几种功能(以Kafka1.0.0版本为准):客户端根据方法的调用创建相应的协议请求,比如创建Topic的createTopics方法,其内部就是发送CreateTopicRequest请求。

能与原理介绍

在Kafka官网中这么描述AdminClient:The AdminClient API supports managing and inspecting topics, brokers, acls, and other Kafka objects. 具体的KafkaAdminClient包含了一下几种功能(以Kafka1.0.0版本为准):

  1. 创建Topic:createTopics(Collection<NewTopic> newTopics)
  2. 删除Topic:deleteTopics(Collection<String> topics)
  3. 罗列所有Topic:listTopics()
  4. 查询Topic:describeTopics(Collection<String> topicNames)
  5. 查询集群信息:describeCluster()
  6. 查询ACL信息:describeAcls(AclBindingFilter filter)
  7. 创建ACL信息:createAcls(Collection<AclBinding> acls)
  8. 删除ACL信息:deleteAcls(Collection<AclBindingFilter> filters)
  9. 查询配置信息:describeConfigs(Collection<ConfigResource> resources)
  10. 修改配置信息:alterConfigs(Map<ConfigResource, Config> configs)
  11. 修改副本的日志目录:alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment)
  12. 查询节点的日志目录信息:describeLogDirs(Collection<Integer> brokers)
  13. 查询副本的日志目录信息:describeReplicaLogDirs(Collection<TopicPartitionReplica> replicas)
  14. 增加分区:createPartitions(Map<String, NewPartitions> newPartitions)
  15. 其内部原理是使用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();
    }
}
Spring Boot集群管理工具KafkaAdminClient

以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

操作系统

操作系统

William Stallings / 陈向群、陈渝 / 机械工业出版社 / 2010.9 / 69.00元

本书不仅全面地讲述了操作系统的基本概念、原理和方法,还清楚地展现了当代操作系统的本质和特点。作者针对近几年操作系统领域的最新变化,对操作系统的设计原理进行深入的阐述,同时将其对操作系统整个领域全面而深入的理解呈现给读者。 本书特色 ·选择Windows Vista、UNIX和Linux三个操作系统作为示例,以帮助读者熟悉当代操作系统的设计原理和实现问题。 ·新增嵌入式操作系统一章......一起来看看 《操作系统》 这本书的介绍吧!

HTML 编码/解码
HTML 编码/解码

HTML 编码/解码

RGB HSV 转换
RGB HSV 转换

RGB HSV 互转工具

HEX HSV 转换工具
HEX HSV 转换工具

HEX HSV 互换工具