Kafka学习之路（五）Kafka在zookeeper中的存储

￡神魔★判官ぃ 2023-01-16 02:56 141阅读 0赞

**目录**

一、Kafka在zookeeper中存储结构图

二、分析

2.1　topic注册信息

2.2　partition状态信息

2.3　Broker注册信息

2.4　Controller epoch

2.5　Controller注册信息

2.6　补充Consumer and Consumer group

2.7　Consumer均衡算法

2.8　Consumer注册信息

2.9　Consumer owner

2.10　Consumer offset

2.11　Re-assign partitions

2.12　Preferred replication election

2.13　删除topics

2.14　Topic配置

--------------------

## 一、Kafka在zookeeper中存储结构图 ##

![c3c579eb440e666b041d755f78734b24.png][]

[回到顶部][Link 1]

## 二、分析 ##

### 2.1　topic注册信息 ###

/brokers/topics/\[topic\] :

存储某个topic的partitions所有分配信息

[zk: localhost:2181(CONNECTED) 1] get /brokers/topics/topic2

![ab1f891b2535f1cf1a347469aa83c28b.png][]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

Schema:
    {
        "version": "版本编号目前固定为数字1",
        "partitions": {
            "partitionId编号": [
                同步副本组brokerId列表
            ],
            "partitionId编号": [
                同步副本组brokerId列表
            ],
            .......
        }
    }
    Example:
    {
    "version": 1,
    "partitions": {
    "2": [1, 2, 3],
    "1": [0, 1, 2],
    "0": [3, 0, 1],
    }
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

### 2.2　partition状态信息 ###

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

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

![3cfdc268520d001404762d8fc889d602.png][]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

Schema:
    {
    "controller_epoch": 表示kafka集群中的中央控制器选举次数,
    "leader": 表示该partition选举leader的brokerId,
    "version": 版本编号默认为1,
    "leader_epoch": 该partition leader选举次数,
    "isr": [同步副本组brokerId列表]
    }
     
    Example:
    {
    "controller_epoch": 1,
    "leader": 3,
    "version": 1,
    "leader_epoch": 0,
    "isr": [3, 0, 1]
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

### 2.3　Broker注册信息 ###

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

每个broker的配置文件中都需要指定一个数字类型的id(全局不可重复),此节点为临时znode(EPHEMERAL)

![7f239a3aa694ec551b618e777d675620.png][]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

Schema:
    {
    "jmx_port": jmx端口号,
    "timestamp": kafka broker初始启动时的时间戳,
    "host": 主机名或ip地址,
    "version": 版本编号默认为1,
    "port": kafka broker的服务端端口号,由server.properties中参数port确定
    }
     
    Example:
    {
    "jmx_port": -1,
    "timestamp":"1525741823119"
    "version": 1,
    "host": "hadoop1",
    "port": 9092
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

### 2.4　Controller epoch ###

/controller\_epoch -->  int (epoch)

此值为一个数字,kafka集群中第一个broker第一次启动时为1，以后只要集群中center controller中央控制器所在broker变更或挂掉，就会重新选举新的center controller，每次center controller变更controller\_epoch值就会 + 1;

![d4c9fb906bad139d72fb1eac272e21fa.png][]

### 2.5　Controller注册信息 ###

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

![b47c7fd34cd0957fe72bc7aace769ae4.png][]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

Schema:
    {
    "version": 版本编号默认为1,
    "brokerid": kafka集群中broker唯一编号,
    "timestamp": kafka broker中央控制器变更时的时间戳
    }
     
    Example:
    {
    "version": 1,
    "brokerid": 0,
    "timestamp": "1525741822769"
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

### 2.6　补充Consumer and Consumer group ###

![b26b5b448bb64fc84e2af47662b8da03.png][]

**a.每个consumer客户端被创建时,会向zookeeper注册自己的信息;**  
**b.此作用主要是为了"负载均衡".**  
**c.同一个Consumer Group中的Consumers，Kafka将相应Topic中的每个消息只发送给其中一个Consumer。**  
**d.Consumer Group中的每个Consumer读取Topic的一个或多个Partitions，并且是唯一的Consumer；**  
**e.一个Consumer group的多个consumer的所有线程依次有序地消费一个topic的所有partitions,如果Consumer group中所有consumer总线程大于partitions数量，则会出现空闲情况;**

> **举例说明：**

> **kafka集群中创建一个topic为report-log   4 partitions 索引编号为0,1,2,3**

> **假如有目前有三个消费者node：注意-->一个consumer中一个消费线程可以消费一个或多个partition.**

> **如果每个consumer创建一个consumer thread线程,各个node消费情况如下，node1消费索引编号为0,1分区，node2费索引编号为2,node3费索引编号为3**

> **如果每个consumer创建2个consumer thread线程，各个node消费情况如下(是从consumer node先后启动状态来确定的)，node1消费索引编号为0,1分区；node2费索引编号为2,3；node3为空闲状态**

**总结：**  
**从以上可知，Consumer Group中各个consumer是根据先后启动的顺序有序消费一个topic的所有partitions的。**

**如果Consumer Group中所有consumer的总线程数大于partitions数量，则可能consumer thread或consumer会出现空闲状态**。

### 2.7　Consumer均衡算法 ###

**当一个group中,有consumer加入或者离开时,会触发partitions均衡.均衡的最终目的,是提升topic的并发消费能力.**  
**1) 假如topic1,具有如下partitions: P0,P1,P2,P3**  
**2) 加入group中,有如下consumer: C0,C1**  
**3) 首先根据partition索引号对partitions排序: P0,P1,P2,P3**  
**4) 根据(consumer.id + '-'+ thread序号)排序: C0,C1**  
**5) 计算倍数: M = \[P0,P1,P2,P3\].size / \[C0,C1\].size,本例值M=2(向上取整)**  
**6) 然后依次分配partitions: C0 = \[P0,P1\],C1=\[P2,P3\],即Ci = \[P(i \* M),P((i + 1) \* M -1)\]**

### 2.8　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;

[zk: localhost:2181(CONNECTED) 11] get /consumers/console-consumer-2304/ids/console-consumer-2304_hadoop2-1525747915241-6b48ff32

![d58f6e9acd314bfd648192d5b1543dd7.png][]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

Schema:
    {
    "version": 版本编号默认为1,
    "subscription": { //订阅topic列表
    "topic名称": consumer中topic消费者线程数
    },
    "pattern": "static",
    "timestamp": "consumer启动时的时间戳"
    }
     
    Example:
    {
    "version": 1,
    "subscription": {
    "topic2": 1
    },
    "pattern": "white_list",
    "timestamp": "1525747915336"
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

### 2.9　Consumer owner ###

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

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.

![735c2cd70f0b6ab08136e83626851583.png][]

### 2.10　Consumer offset ###

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

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

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

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

![ac4910347d35ef5c64ade3743af9a852.png][]

### 2.11　Re-assign partitions ###

/admin/reassign\_partitions

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

{
       "fields":[
          {
             "name":"version",
             "type":"int",
             "doc":"version id"
          },
          {
             "name":"partitions",
             "type":{
                "type":"array",
                "items":{
                   "fields":[
                      {
                         "name":"topic",
                         "type":"string",
                         "doc":"topic of the partition to be reassigned"
                      },
                      {
                         "name":"partition",
                         "type":"int",
                         "doc":"the partition to be reassigned"
                      },
                      {
                         "name":"replicas",
                         "type":"array",
                         "items":"int",
                         "doc":"a list of replica ids"
                      }
                   ],
                }
                "doc":"an array of partitions to be reassigned to new replicas"
             }
          }
       ]
    }
     
    Example:
    {
      "version": 1,
      "partitions":
         [
            {
                "topic": "Foo",
                "partition": 1,
                "replicas": [0, 1, 3]
            }
         ]            
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

### 2.12　Preferred replication election ###

/admin/preferred\_replica\_election

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

{
       "fields":[
          {
             "name":"version",
             "type":"int",
             "doc":"version id"
          },
          {
             "name":"partitions",
             "type":{
                "type":"array",
                "items":{
                   "fields":[
                      {
                         "name":"topic",
                         "type":"string",
                         "doc":"topic of the partition for which preferred replica election should be triggered"
                      },
                      {
                         "name":"partition",
                         "type":"int",
                         "doc":"the partition for which preferred replica election should be triggered"
                      }
                   ],
                }
                "doc":"an array of partitions for which preferred replica election should be triggered"
             }
          }
       ]
    }
     
    例子:
     
    {
      "version": 1,
      "partitions":
         [
            {
                "topic": "Foo",
                "partition": 1         
            },
            {
                "topic": "Bar",
                "partition": 0         
            }
         ]            
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

### 2.13　删除topics ###

/admin/delete\_topics

![293fc457e98a69f920e76516ee87df75.png][]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

Schema:
    { "fields":
        [ {"name": "version", "type": "int", "doc": "version id"},
          {"name": "topics",
           "type": { "type": "array", "items": "string", "doc": "an array of topics to be deleted"}
          } ]
    }
     
    例子:
    {
      "version": 1,
      "topics": ["foo", "bar"]
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

### 2.14　Topic配置 ###

/config/topics/\[topic\_name\]

![650b80943155bffa409d89b32a21969a.png][]

[c3c579eb440e666b041d755f78734b24.png]: /images/20221022/89ee0a1568844c37a4f071703a8fe654.png
[Link 1]: https://www.cnblogs.com/qingyunzong/p/9007107.html#_labelTop
[ab1f891b2535f1cf1a347469aa83c28b.png]: /images/20221022/1aaa7597ccbb43a7885a2abd640bac35.png
[48304ba5e6f9fe08f3fa1abda7d326ab.png]: /images/20221022/62d8858ddf8b498083858fed0193382a.png
[3cfdc268520d001404762d8fc889d602.png]: /images/20221022/d542d5b5c56a41fcaa69d02150ba7917.png
[7f239a3aa694ec551b618e777d675620.png]: /images/20221022/0f970e2fb70e402099a56962b736fabe.png
[d4c9fb906bad139d72fb1eac272e21fa.png]: /images/20221022/321d84c1eabc41859fb511181b8a9ebf.png
[b47c7fd34cd0957fe72bc7aace769ae4.png]: /images/20221022/3a63ce26ca944737b7e0c71ce59c5efa.png
[b26b5b448bb64fc84e2af47662b8da03.png]: /images/20221022/88f40bb84f7147f7819a27c71e8032b8.png
[d58f6e9acd314bfd648192d5b1543dd7.png]: /images/20221022/2124b8dd02a84113909c640753438c62.png
[735c2cd70f0b6ab08136e83626851583.png]: /images/20221022/0aab03151d2545b1b39fbf0158249a7b.png
[ac4910347d35ef5c64ade3743af9a852.png]: /images/20221022/352143b0282d4636bbe41efb6797d655.png
[293fc457e98a69f920e76516ee87df75.png]: /images/20221022/6c33ad0a1cc2425d84c16b40ded52361.png
[650b80943155bffa409d89b32a21969a.png]: /images/20221022/b2d5d30c5b4947d1bb4a0191b7779c7f.png