kafka在zookeeper中的存储结构_H5之家 - 中国HTML5教程资源分享第一站

> WEB开发 > JavaScript > 正文

编程大巴

kafka在zookeeper中的存储结构

作者：发布日期：2016-01-13 23:14:16

Tag标签：结构

参考site：#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':'h3', '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' }

如图：

8. consumer owner

/consumers/[groupId]/owners/[topic]/[partitionId] -> consumerIdString + threadId索引编号

当consumer启动时,所触发的操作:

a) 首先进行'Consumer Id注册';

b) 然后在'Consumer id 注册'节点下注册一个watch用来监听当前group中其他consumer的'退出'和'加入';只要此znode path下节点列表变更,

都会触发此group下consumer的负载均衡.(比如一个consumer失效,那么其他consumer接管partitions).

c) 在'Broker id 注册'节点下,注册一个watch用来监听broker的存活情况;如果broker列表变更,将会触发所有的groups下的consumer重新balance.

9. consumer offset

/consumers/[groupId]/offsets/[topic]/[partitionId] -> long (offset)

用来跟踪每个consumer目前所消费的partition中最大的offset

此znode为持久节点,可以看出offset跟group_id有关,以表明当消费者组(consumer group)中一个消费者失效,

重新触发balance,其他consumer可以继续消费.

10. topic 配置

/config/topics/[topic_name]