查看kafka集群多少几点

Kafka集群是一个分布式系统，它由多个Kafka Broker组成。每个Kafka Broker都可以管理多个Kafka Topic的分区，所以一个Kafka集群可以具有非常高的扩展性和可伸缩性。但是，对于管理Kafka集群，我们需要了解Kafka集群中的Broker数量、各个Broker的状态、各个Broker所负责的分区等关键信息。本文将介绍如何查看Kafka集群的一些关键信息。

1. 查看Kafka集群中Broker的数量

可以使用Kafka提供的Java API来获取Kafka集群中的Broker数量。下面是一个使用Java API的示例代码：

import kafka.cluster.Broker;
import kafka.utils.ZkUtils;
import java.util.List;

public class KafkaClusterMetrics {

    public static void main(String[] args) {
        String zookeeperConnect = "localhost:2181"; // ZooKeeper连接配置
        int sessionTimeoutMs = 10000; // 会话超时时间
        int connectionTimeoutMs = 10000; // 连接超时时间
        ZkUtils zkUtils = ZkUtils.apply(zookeeperConnect, sessionTimeoutMs, connectionTimeoutMs, false);
        List<Broker> brokers = scala.collection.JavaConversions.seqAsJavaList(zkUtils.getAllBrokersInCluster());
        int brokerCount = brokers.size();
        System.out.println("Kafka cluster contains " + brokerCount + " brokers.");
    }

}

2. 查看Kafka集群中各个Broker的状态

可以使用Kafka提供的Java API来获取Kafka集群中各个Broker的状态。下面是一个使用Java API的示例代码：

import kafka.cluster.Broker;
import kafka.utils.ZkUtils;
import scala.Option;

public class KafkaClusterMetrics {

    public static void main(String[] args) {
        String zookeeperConnect = "localhost:2181"; // ZooKeeper连接配置
        int sessionTimeoutMs = 10000; // 会话超时时间
        int connectionTimeoutMs = 10000; // 连接超时时间
        ZkUtils zkUtils = ZkUtils.apply(zookeeperConnect, sessionTimeoutMs, connectionTimeoutMs, false);
        List<Broker> brokers = scala.collection.JavaConversions.seqAsJavaList(zkUtils.getAllBrokersInCluster());
        for (Broker broker : brokers) {
            Option<Object> leaderOption = zkUtils.getLeaderForBroker(broker.id());
            if (leaderOption.isDefined()) {
                System.out.println("Broker " + broker.id() + " is " + (