Kafka Admin 是一個用于管理 Apache Kafka 集群的工具。它提供了一系列操作,如創建主題、刪除主題、列出主題、修改分區數等。要使用 Kafka Admin,你需要遵循以下步驟:
首先,你需要將 Kafka Admin 客戶端庫添加到你的項目中。如果你使用的是 Maven,可以在 pom.xml
文件中添加以下依賴:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-admin-client</artifactId>
<version>3.0.0</version>
</dependency>
對于 Gradle 項目,在 build.gradle
文件中添加以下依賴:
implementation 'org.apache.kafka:kafka-admin-client:3.0.0'
請注意,版本號可能會有所不同,請根據你使用的 Kafka 版本進行調整。
在你的 Java 代碼中,創建一個 KafkaAdmin
實例,指定 Kafka 集群的bootstrap servers 地址:
Properties adminClientProps = new Properties();
adminClientProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
AdminClient adminClient = AdminClient.create(adminClientProps);
現在你可以使用 KafkaAdmin
實例執行各種操作。例如,要創建一個名為 my_topic
的主題,具有 3 個分區和一個副本因子,可以執行以下操作:
NewTopic newTopic = new NewTopic("my_topic", 3, (short) 1);
CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(newTopic));
createTopicsResult.all().get(); // 等待操作完成
要刪除一個名為 my_topic
的主題,可以執行以下操作:
DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Collections.singletonList("my_topic"));
deleteTopicsResult.all().get(); // 等待操作完成
要列出現有的主題,可以執行以下操作:
ListTopicsResult listTopicsResult = adminClient.listTopics();
Set<String> topics = listTopicsResult.names().get();
topics.forEach(System.out::println);
在完成所有操作后,確保關閉 KafkaAdmin
實例以釋放資源:
adminClient.close();
這就是如何使用 Kafka Admin 進行基本操作。你可以根據實際需求執行其他操作,如修改分區數、描述主題配置等。更多詳細信息和示例,請參考 Kafka Admin 文檔。