要獲取Kafka生產者的元數據,可以使用Kafka提供的AdminClient API。以下是一個示例代碼:
import java.util.Properties;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.clients.admin.ListConsumerGroupsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.TopicListing;
public class KafkaProducerMetadata {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); // Kafka集群的地址
try (AdminClient adminClient = AdminClient.create(props)) {
// 獲取所有主題的元數據
ListTopicsResult topicsResult = adminClient.listTopics();
topicsResult.listings().get().forEach(topicListing -> {
System.out.println("Topic: " + topicListing.name());
});
// 獲取所有消費者組的元數據
ListConsumerGroupsResult consumerGroupsResult = adminClient.listConsumerGroups();
consumerGroupsResult.all().get().forEach(consumerGroupListing -> {
System.out.println("Consumer Group: " + consumerGroupListing.groupId());
});
} catch (Exception e) {
e.printStackTrace();
}
}
}
上述代碼使用AdminClient API創建一個AdminClient實例,并使用該實例來獲取所有主題和消費者組的元數據。可以根據需要進行修改和擴展。