本文共 4551 字，大约阅读时间需要 15 分钟。

Kafka AdminClient API入门及常用操作示例

1. Kafka客户端API类型

在Kafka中，客户端API主要分为以下几类，各自承担不同的功能：

• AdminClient API：用于管理和检测Topic、Broker及其他Kafka相关对象。
• Producer API：允许将消息发布到一个或多个Topic。
• Consumer API：支持订阅一个或多个Topic并处理生成的消息。
• Streams API：用于将输入流高效转换为输出流。
• Connector API：从源系统或应用程序拉取数据到Kafka。

2. AdminClient API功能概述

AdminClient是Kafka中用于管理Topic和Broker等组件的核心API，常见操作如下：

2.1 创建Topic

要创建一个新的Topic，可以按照以下步骤操作：

Short replicationFactor = 1;
Topictyping newTopic = new NewTopic topicsName, replicationFactor, PARTITION_NUM);
CreateTopicsResult createTopicResult = adminClient.createTopics(Arrays.asList(newTopic));

2.2 查询Topic列表

要列出所有存在的Topic，可以执行以下操作：

ListTopicsResult listTopicsResult = adminClient.listTopics();
Set
   
     topicNames = listTopicsResult.names().get();
topicNames.forEach(topic -> {
    System.out.println("topic: " + topic);
});
   

2.3 删除Topic

要删除一个Topic，可以使用以下代码：

DeleteTopicsResult deleteTopicResult = adminClient.deleteTopics(Arrays.asList(topicName));

2.4 描述Topic详细信息

要获取某个Topic的详细信息，可以执行描述Topic操作：

DescribeTopicsResult describeTopicResult = adminClient.describeTopics(Arrays.asList(topicName));
Map
   
     topicDescMap = describeTopicResult.all().get();
topicDescMap.forEach((topicName, description) {
    System.out.println("topicName: " + topicName + ", description: " + description);
});
   

2.5 修改Topic配置

要修改Topic的配置，可以按照以下方法操作：

AlterConfigsResult alterConfigResult = adminClient.alterConfigs(configMaps);

3. 代码实现示例

在实际使用中，可以将上述操作整合到代码中。以下是一个完整的代码示例：

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.AlterConfigsResult;
import org.apache.kafka.common.KafkaFuture;
import java.util.Arrays;
import java.util.Collections;
@Slf4j
public class AdminSample {
    private static final String TOPIC_NAME = "demo_topic";
    public static void main(String[] args) throws Exception {
        AdminClient adminClient = getAdminClient();
        
        // 创建Topic
        createTopic(adminClient);
        // 查询Topic列表
        ListTopics(adminClient);
        // 删除Topic
        DeleteTopic(adminClient);
        // 描述Topic详情
        DescribeTopicDetails(adminClient);
        // 查询配置信息
        DescribeConfigs(adminClient);
        // 修改配置信息
        AlterConfigs(adminClient);
        // 增加分区数
        createPartitions(adminClient);
    }
    private static AdminClient getAdminClient() {
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        return AdminClient.create(properties);
    }
    private static void createTopic(AdminClient adminClient) {
        Short replicationFactor = 1;
        NewTopic newTopic = new NewTopic(TOPIC_NAME, 1, replicationFactor);
        CreateTopicsResult createTopicResult = adminClient.createTopics(Arrays.asList(newTopic));
        log.info("创建Topic {}成功!", TOPIC_NAME);
    }
    private static void ListTopics(AdminClient adminClient) {
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        Set
   
     topicNames = listTopicsResult.names().get();
        topicNames.forEach(topic -> log.info("发现Topic: {}", topic));
    }
    private static void DeleteTopic(AdminClient adminClient) {
        DeleteTopicsResult deleteResult = adminClient.deleteTopics(Arrays.asList(TOPIC_NAME));
        log.info("删除Topic {}成功!", TOPIC_NAME);
    }
    private static void DescribeTopicDetails(AdminClient adminClient) {
        DescribeTopicsResult describeResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME));
        Map
    
      topicDetails = describeResult.all().get();
        topicDetails.forEach((topicName, detail) -> {
            log.info("Topic {}的详情：{}", topicName, detail);
        });
    }
    // ... （其他操作如DescribeConfigs、AlterConfigs、createPartitions等）
}
    
   

4. pom.xml配置

确保项目中已引入必要的依赖项：

   
    
    
        
     
      org.springframework.boot
     
        
     
      spring-boot-starter
     
    
    
    
    
        
     
      org.springframework.boot
     
        
     
      spring-boot-starter-web
     
    
    
    
    
        
     
      org.project.lombok
     
        
     
      lombok
     
    
    
    
    
        
     
      org.apache.kafka
     
        
     
      kafka-clients
     
    
    
   

5.注意事项

• 配置管理：尽量使用 centroid-based 或 balanced 分区分配策略。
• 权限控制：根据实际需求设置相应的权限。
• 监控与日志：建议开启监控，关注Topic和Broker的运行状态。
• 故障排除：遇到问题时，可参考Kafka的官方文档或者相关社区资源。