kafka在zookeeper中的存储结构-白红宇

kafka在zookeeper中的存储结构

阅读量：4656 次

发布时间：2019-06-09

本文共 3806 字，大约阅读时间需要 12 分钟。

参考site：

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"}

如图：