图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Kafka › 介绍解说› kafka在zookeeper中存储结构
分享

kafka在zookeeper中存储结构

howtodown 2014-11-11 15:45:17 发表于 介绍解说 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 1 29889
问题导读





1.kafka是如何在zookeeper中存储的？
2.什么是Consumer and Consumer group？







1.topic注册信息
/brokers/topics/[topic] :
存储某个topic的partitions所有分配信息
  1. 

  2. Schema:

  3. {

  4.     "version": "版本编号目前固定为数字1",

  5.     "partitions": {

  6.         "partitionId编号": [

  7.             同步副本组brokerId列表

  8.         ],

  9.         "partitionId编号": [

  10.             同步副本组brokerId列表

  11.         ],

  12.         .......

  13.     }

  14. }

  15. Example:

  16. {

  17. "version": 1,

  18. "partitions": {

  19. "0": [1, 2],

  20. "1": [2, 1],

  21. "2": [1, 2],

  22. }

  23. }

  24. 说明：紫红色为patitions编号，蓝色为同步副本组brokerId列表
复制代码


2.partition状态信息
/brokers/topics/[topic]/partitions/[0...N]  其中[0..N]表示partition索引号
/brokers/topics/[topic]/partitions/[partitionId]/state
  1. 

  2. Schema:

  3. {

  4. "controller_epoch": 表示kafka集群中的中央控制器选举次数,

  5. "leader": 表示该partition选举leader的brokerId,

  6. "version": 版本编号默认为1,

  7. "leader_epoch": 该partition leader选举次数,

  8. "isr": [同步副本组brokerId列表]

  9. }

  10. 

  11. Example:

  12. {

  13. "controller_epoch": 1,

  14. "leader": 2,

  15. "version": 1,

  16. "leader_epoch": 0,

  17. "isr": [2, 1]

  18. }
复制代码


3. Broker注册信息
/brokers/ids/[0...N]                 
每个broker的配置文件中都需要指定一个数字类型的id(全局不可重复),此节点为临时znode(EPHEMERAL)
  1. 

  2. 

  3. Schema:

  4. {

  5. "jmx_port": jmx端口号,

  6. "timestamp": kafka broker初始启动时的时间戳,

  7. "host": 主机名或ip地址,

  8. "version": 版本编号默认为1,

  9. "port": kafka broker的服务端端口号,由server.properties中参数port确定

  10. }

  11. 

  12. Example:

  13. {

  14. "jmx_port": 6061,

  15. "timestamp":"1403061899859"

  16. "version": 1,

  17. "host": "192.168.1.148",

  18. "port": 9092

  19. }
复制代码


4. Controller epoch:
/controller_epoch -> int (epoch)  
此值为一个数字,kafka集群中第一个broker第一次启动时为1，以后只要集群中center controller中央控制器所在broker变更或挂掉，就会重新选举新的center controller，每次center
controller变更controller_epoch值就会 + 1;

5. Controller注册信息:
/controller -> int (broker id of the controller)  存储center controller中央控制器所在kafka broker的信息
  1. 

  2. 

  3. Schema:

  4. {

  5. "version": 版本编号默认为1,

  6. "brokerid": 

  7. kafka集群中broker唯一编号,

  8. "timestamp": kafka broker中央控制器变更时的时间戳

  9. }

  10. 

  11. Example:

  12. {

  13. "version": 1,

  14. "brokerid": 3,

  15. "timestamp": "1403061802981"

  16. }
复制代码
[/table]
Consumer and Consumer group概念:



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会出现空闲状态。


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)]


6. 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;
  1. 

  2. Schema:

  3. {

  4. "version": 

  5. 版本编号默认为1,

  6. "subscription": { //订阅topic列表

  7. "topic名称": consumer中topic消费者线程数

  8. },

  9. "pattern": "static",

  10. "timestamp": "consumer启动时的时间戳"

  11. }

  12. 

  13. Example:

  14. {

  15. "version": 1,

  16. "subscription": {

  17. "open_platform_opt_push_plus1": 5

  18. },

  19. "pattern": "static",

  20. "timestamp": "1411294187842"

  21. }
复制代码



7. 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.

8. Consumer offset:
/consumers/[groupId]/offsets/[topic]/[partitionId] -> long (offset)
用来跟踪每个consumer目前所消费的partition中最大的offset
此znode为持久节点,可以看出offset跟group_id有关,以表明当消费者组(consumer group)中一个消费者失效,
重新触发balance,其他consumer可以继续消费.

9. Re-assign partitions
/admin/reassign_partitions
  1. 

  2. 

  3. {

  4.    "fields":[

  5.       {

  6.          "name":"version",

  7.          "type":"int",

  8.          "doc":"version id"

  9.       },

  10.       {

  11.          "name":"partitions",

  12.          "type":{

  13.             "type":"array",

  14.             "items":{

  15.                "fields":[

  16.                   {

  17.                      "name":"topic",

  18.                      "type":"string",

  19.                      "doc":"topic

  20. of the partition to be reassigned"

  21.                   },

  22.                   {

  23.                      "name":"partition",

  24.                      "type":"int",

  25.                      "doc":"the

  26. partition to be reassigned"

  27.                   },

  28.                   {

  29.                      "name":"replicas",

  30.                      "type":"array",

  31.                      "items":"int",

  32.                      "doc":"a

  33. list of replica ids"

  34.                   }

  35.                ],

  36.             }

  37.             "doc":"an array

  38. of partitions to be reassigned to new replicas"

  39.          }

  40.       }

  41.    ]

  42. }

  43. 

  44. Example:

  45. {

  46.   "version": 1,

  47.   "partitions":

  48.      [

  49.         {

  50.             "topic": "Foo",

  51.             "partition": 1,

  52.             "replicas": [0, 1, 3]

  53.         }

  54.      ]            

  55. }
复制代码
10. Preferred replication election
  1. {

  2.    "fields":[

  3.       {

  4.          "name":"version",

  5.          "type":"int",

  6.          "doc":"version id"

  7.       },

  8.       {

  9.          "name":"partitions",

  10.          "type":{

  11.             "type":"array",

  12.             "items":{

  13.                "fields":[

  14.                   {

  15.                      "name":"topic",

  16.                      "type":"string",

  17.                      "doc":"topic

  18. of the partition for which preferred replica election should be triggered"

  19.                   },

  20.                   {

  21.                      "name":"partition",

  22.                      "type":"int",

  23.                      "doc":"the

  24. partition for which preferred replica election should be triggered"

  25.                   }

  26.                ],

  27.             }

  28.             "doc":"an array

  29. of partitions for which preferred replica election should be triggered"

  30.          }

  31.       }

  32.    ]

  33. }

  34. 

  35. 例子:

  36. 

  37. {

  38.   "version": 1,

  39.   "partitions":

  40.      [

  41.         {

  42.             "topic": "Foo",

  43.             "partition": 1         

  44.         },

  45.         {

  46.             "topic": "Bar",

  47.             "partition": 0         

  48.         }

  49.      ]            

  50. }
复制代码
11. 删除topics
/admin/delete_topics
  1. 

  2. Schema:

  3. { "fields":

  4.     [ {"name": "version", "type": "int", "doc": "version

  5. id"},

  6.       {"name": "topics",

  7.        "type": { "type": "array", "items": "string", "doc": "an

  8. array of topics to be deleted"}

  9.       } ]

  10. }

  11. 

  12. 例子:

  13. {

  14.   "version": 1,

  15.   "topics": ["foo", "bar"]

  16. }
复制代码
Topic配置
/config/topics/[topic_name]
例子
  1. 

  2. 

  3. {

  4.   "version": 1,

  5.   "config": {

  6.     "config.a": "x",

  7.     "config.b": "y",

  8.     ...

  9.   }

  10. }
复制代码
[table]

请注明转载自:http://blog.csdn.net/lizhitao/article/details/23744675

回复

使用道具 举报

自定义广告语

已有(1)人评论

跳转到指定楼层
正序浏览
loreting 发表于 2014-11-11 16:09:36
提示: 作者被禁止或删除 内容自动屏蔽
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
howtodown

版主

关注

  • 666

    主题

  • 1249

    帖子

  • 322

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000