kafka在zookeeper中的存儲結構

本文轉載自查看原文 2016-01-13 15:22 8905 kafka運維

參考site：http://kafka.apache.org/documentation.html#impl_zookeeper

1、zookeeper客戶端相關命令

在確保zookeeper服務啟動狀態下，通過 bin/zkCli.sh -server 127.0.0.1:2181 該命令來連接客戶端

簡單操作如下：

　　1. 顯示根目錄下、文件： ls / 使用 ls 命令來查看當前 ZooKeeper 中所包含的內容

　　2. 顯示根目錄下、文件： ls2 / 查看當前節點數據並能看到更新次數等數據

　　3. 創建文件，並設置初始內容： create /zk "test" 創建一個新的 znode節點“ zk ”以及與它關聯的字符串

　　4. 獲取文件內容： get /zk 確認 znode 是否包含我們所創建的字符串

　　5. 修改文件內容： set /zk "zkbak" 對 zk 所關聯的字符串進行設置

　　6. 刪除文件： delete /zk 將剛才創建的 znode 刪除

　　7. 退出客戶端： quit

　　8. 幫助命令： help

2、topic注冊信息

/brokers/topics/[topic] :

存儲某個topic的partitions所有分配信息

Schema:
{
    "version": "版本編號目前固定為數字1",
    "partitions": {
        "partitionId編號": [
            同步副本組brokerId列表
        ],
        "partitionId編號": [
            同步副本組brokerId列表
        ],
        .......
    }
}
Example:
{
    "version": 1,
    "partitions": {
	"0":[0,1,2]
     }
}

如下圖：

3.partition狀態信息

/brokers/topics/[topic]/partitions/[0...N] 其中[0..N]表示partition索引號

/brokers/topics/[topic]/partitions/[partitionId]/state

Schema:
{
    "controller_epoch": 表示kafka集群中的中央控制器選舉次數,
    "leader": 表示該partition選舉leader的brokerId,
    "version": 版本編號默認為1,
    "leader_epoch": 該partition leader選舉次數,
    "isr": [同步副本組brokerId列表]
}
 
Example:
{
    "controller_epoch":20,
    "leader":0,
    "version":1,
    "leader_epoch":0,
    "isr":[0,1,2]
}

如圖：

4. broker注冊信息

/brokers/ids/[0...N]

每個broker的配置文件中都需要指定一個數字類型的id(全局不可重復),此節點為臨時znode(EPHEMERAL)

Schema:
{
    "jmx_port": jmx端口號,
    "timestamp": kafka broker初始啟動時的時間戳,
    "host": 主機名或ip地址,
    "version": 版本編號默認為1,
    "port": kafka broker的服務端端口號,由server.properties中參數port確定
}
 
Example:
{
    "jmx_port":1,
    "timestamp":"1452068227537",
    "host":"h1",
    "version":1,
    "port":9092
}

如圖：

5. controller epoch

/controller_epoch -> int (epoch)

此值為一個數字,kafka集群中第一個broker第一次啟動時為1，以后只要集群中center controller中央控制器所在broker變更或掛掉，就會重新選舉新的center controller，每次center controller變更controller_epoch值就會 + 1;

如圖：

6. controller注冊信息

/controller -> int (broker id of the controller) 存儲center controller中央控制器所在kafka broker的信息

Schema:
{
    "version": 版本編號默認為1,
    "brokerid": kafka集群中broker唯一編號,
    "timestamp": kafka broker中央控制器變更時的時間戳
}
 
Example:
{
    "version":1,
    "brokerid":0,
    "timestamp":"1452068227409"
}

如圖：

7. consumer注冊信息

每個consumer都有一個唯一的ID(consumerId可以通過配置文件指定,也可以由系統生成),此id用來標記消費者信息.

/consumers/[groupId]/ids/[consumerIdString]

是一個臨時的znode,此節點的值為請看consumerIdString產生規則,即表示此consumer目前所消費的topic + partitions列表.

consumerId產生規則： 

   StringconsumerUuid = null; 
       if(config.consumerId!=null && config.consumerId){ 
           consumerUuid = consumerId; 
       }else { 
           String uuid = UUID.randomUUID() 
           consumerUuid = "%s-%d-%s".format( 
                InetAddress.getLocalHost.getHostName, System.currentTimeMillis, 
                uuid.getMostSignificantBits().toHexString.substring(0,8)); 

       } 
  String consumerIdString = config.groupId + "_" + consumerUuid;  

Schema:
{
    "version": 版本編號默認為1,
    "subscription": { //訂閱topic列表
        "topic名稱": consumer中topic消費者線程數
    },
    "pattern": "static",
    "timestamp": "consumer啟動時的時間戳"
}
 
Example:
{
    "version":1,
    "subscription":{
        "replicatedtopic":1
    },
    "pattern":"white_list",
    "timestamp":"1452134230082"
}

如圖：