Kafka静态IP和服务发现

Kafka中的静态IP和服务发现是构建分布式系统的重要组成部分。Kafka集群由多个实例组成，每个实例都必须公开一个稳定的IP地址和端口号，以便其他实例可以找到它并建立通信。同时，服务发现还可以用于自动发现Kafka实例。以下是使用ZooKeeper进行Kafka服务发现的示例代码：

public class KafkaBrokerZkPath {
  private static String zkPath = "/brokers/ids";

  public static String getBrokerPath(int id) {
    return zkPath + "/" + id;
  }

  public static String getAllBrokersPath() {
    return zkPath;
  }
}

public class KafkaBrokerTopicEventListener {
  private static String zkHost = "localhost:2181";
  private static String topicPath = "/brokers/topics";

  public static void main(String[] args) {
    ZooKeeper zooKeeper = null;
    try {
      zooKeeper = new ZooKeeper(zkHost, 5000, new KafkaBrokerTopicWatcher());
      zooKeeper.exists(topicPath, true);
      Thread.sleep(Integer.MAX_VALUE);
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      if (zooKeeper != null) {
        try {
          zooKeeper.close();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }

  static class KafkaBrokerTopicWatcher implements Watcher {
    @Override
    public void process(WatchedEvent watchedEvent) {
      if (watchedEvent.getType().equals(Event.EventType.NodeChildrenChanged)) {
        ZooKeeper zooKeeper = null;
        try {
          zooKeeper = new ZooKeeper(zkHost, 5000, new KafkaBrokerTopicWatcher());
          List<String> topics = zooKeeper.getChildren(topicPath, true);
          System.out.println("Topics:");
          for (String topic : topics) {
            System.out.println(topic);
            List<String> partitions = zooKeeper.getChildren(topicPath
                + "/" + topic + "/partitions", false);
            System.out.println("\tPartitions:");
            for (String partition : partitions) {
              byte[] bytes = zooKeeper.getData(topicPath + "/"
                  + topic + "/partitions/" + partition, false, null);
              String[] brokerIds = new String(bytes, "UTF-8").split(",");
              System.out.println("\t\tPartition " + partition + ":");
              for (String brokerId : brokerIds) {
                String brokerPath = KafkaBrokerZkPath.getBrokerPath(Integer.parseInt(brokerId));
                byte[] brokerBytes = zooKeeper.getData(brokerPath