美文网首页kafka
慕了，我要是早点看到这篇写 Kafka 的分区管理的文章就好了

慕了，我要是早点看到这篇写 Kafka 的分区管理的文章就好了

作者: 程序花生 | 来源:发表于2021-05-04 16:37 被阅读0次

    Kafka可以将主题划分为多个分区（Partition），会根据分区规则选择把消息存储到哪个分区中，只要如果分区规则设置的合理，那么所有的消息将会被均匀的分布到不同的分区中，这样就实现了负载均衡和水平扩展。另外，多个订阅者可以从一个或者多个分区中同时消费数据，以支撑海量数据处理能力。

    顺便说一句，由于消息是以追加到分区中的，多个分区顺序写磁盘的总效率要比随机写内存还要高（引用Apache Kafka – A High Throughput Distributed Messaging System的观点），是Kafka高吞吐率的重要保证之一。

    一、副本机制

    由于Producer和Consumer都只会与Leader角色的分区副本相连，所以kafka需要以集群的组织形式提供主题下的消息高可用。kafka支持主备复制，所以消息具备高可用和持久性。

    一个分区可以有多个副本，这些副本保存在不同的broker上。每个分区的副本中都会有一个作为Leader。当一个broker失败时，Leader在这台broker上的分区都会变得不可用，kafka会自动移除Leader，再其他副本中选一个作为新的Leader。

    在通常情况下，增加分区可以提供kafka集群的吞吐量。然而，也应该意识到集群的总分区数或是单台服务器上的分区数过多，会增加不可用及延迟的风险。

    itcast@Server-node:/mnt/d/kafka_2.12-2.2.1$ bin/kafka-topics.sh --describe -- 
zookeeper localhost:2181 --topic heima 
// hema这个主题有三个分区，一个副本同处在一个节点当中 
Topic:heima     PartitionCount:3    ReplicationFactor:1     Configs: 
Topic: heima    Partition: 0    Leader: 0   Replicas: 0     Isr: 0 
Topic: heima    Partition: 1    Leader: 0   Replicas: 0     Isr: 0 
Topic: heima    Partition: 2    Leader: 0   Replicas: 0     Isr: 0

    二、分区Leader选举

    可以预见的是,如果某个分区的Leader挂了,那么其它跟随者将会进行选举产生一个新的leader,之后所有的读写就会转移到这个新的Leader上,在kafka中,其不是采用常见的多数选举的方式进行副本的Leader选举,而是会在Zookeeper上针对每个Topic维护一个称为 ISR（in-sync replica，已同步的副本） 的集合,显然还有一些副本没有来得及同步。只有这个ISR列表里面的才有资格成为leader( 先使用ISR里面的第一个，如果不行依次类推，因为ISR里面的是同步副本，消息是最完整且各个节点都是一样的 )。 通过ISR,kafka需要的冗余度较低，可以容忍的失败数比较高。假设某个topic有f+1个副本，kafka可以容忍f个不可用,当然,如果全部ISR里面的副本都不可用,也可以选择其他可用的副本,只是存在数据的不一致。

    三、分区重新分配

    我们往已经部署好的Kafka集群里面添加机器是最正常不过的需求，而且添加起来非常地方便，我们需要做的事是从已经部署好的Kafka节点中复制相应的配置文件，然后把里面的broker id修改成全局唯一的，最后启动这个节点即可将它加入到现有Kafka集群中。

    但是问题来了，新添加的Kafka节点并不会自动地分配数据，所以无法分担集群的负载，除非我们新建一个topic。但是现在我们想手动将部分分区移到新添加的Kafka节点上，Kafka内部提供了相关的工具来重新分布某个topic的分区。

    具体步骤

    • 第一步：我们创建一个有三个节点的集群，详情可查看第九章集群的搭建
    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ bin/kafka-topics.sh --create -- 
zookeeper localhost:2181 --topic heima-par --partitions 3 --replication-factor 3 
Created topic heima-par.

    详情查看

    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ bin/kafka-topics.sh --describe 
--zookeeper localhost:2181 --topic heima 
-par 
Topic:heima-par PartitionCount:3    ReplicationFactor:3     Configs: 
Topic: heima-par    Partition: 0    Leader: 2   Replicas: 2,1,0 Isr: 2,1,0 
Topic: heima-par    Partition: 1    Leader: 0   Replicas: 0,2,1 Isr: 0 
Topic: heima-par    Partition: 2    Leader: 1   Replicas: 1,0,2 Isr: 1,0,2 
itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$

    从上面的输出可以看出heima-par这个主题一共有三个分区，有三个副本

    • 第二步：主题heima-par再添加一个分区
    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ bin/kafka-topics.sh --alter -- zookeeper localhost:2181 --topic heima-pa 
r --partitions 4 
WARNING: If partitions are increased for a topic that has a key, the partition 
logic or ordering of the messages will be affected 
Adding partitions succeeded!

    查看详情已经变成4个分区

    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ bin/kafka-topics.sh --describe 
--zookeeper localhost:2181 --topic heima 
-par 
Topic:heima-par PartitionCount:4    ReplicationFactor:3     Configs: 
Topic: heima-par    Partition: 0    Leader: 2   Replicas: 2,1,0 Isr: 2,1,0 
Topic: heima-par    Partition: 1    Leader: 0   Replicas: 0,2,1 Isr: 0 
Topic: heima-par    Partition: 2    Leader: 1   Replicas: 1,0,2 Isr: 1,0,2 
Topic: heima-par    Partition: 3    Leader: 2   Replicas: 2,1,0 Isr: 2,1,0

    这样会导致broker2维护更多的分区

    • 第三步：再添加一个broker节点

    查看主题信息

    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ bin/kafka-topics.sh --describe 
--zookeeper localhost:2181 --topic heima 
-par 
Topic:heima-par PartitionCount:4    ReplicationFactor:3     Configs: 
Topic: heima-par    Partition: 0    Leader: 2   Replicas: 2,1,0 Isr: 2,1,0 
Topic: heima-par    Partition: 1    Leader: 0   Replicas: 0,2,1 Isr: 0 
Topic: heima-par    Partition: 2    Leader: 1   Replicas: 1,0,2 Isr: 1,0,2 
Topic: heima-par    Partition: 3    Leader: 2   Replicas: 2,1,0 Isr: 2,1,0

    从上面输出信息可以看出新添加的节点并没有分配之前主题的分区

    • 第四步：重新分配

    现在我们需要将原先分布在broker 1-3节点上的分区重新分布到broker 1-4节点上，借助 kafka-reassign-partitions.sh工具生成reassign plan，不过我们先得按照要求定义一个文件，里面说明哪些topic需要重新分区，文件内容如下：

    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ cat reassign.json {
    "topics":[{"topic":"heima-par"}], 
    "version":1 
}

    然后使用 kafka-reassign-partitions.sh 工具生成reassign plan

    然后执行脚本 bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to -move-json-file reassign.json --broker-list "0,1,2,3" --generate

    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ bin/kafka-reassign- 
partitions.sh --zookeeper localhost:2181 --topics-to 
-move-json-file reassign.json --broker-list "0,1,2,3" --generate 
Current partition replica assignment 
{"version":1,"partitions":[{"topic":"heima-par","partition":2,"replicas": [1,0,2],"log_dirs":["any","any","any"]},
{"topic":"heima- par","partition":1,"replicas":[0,2,1],"log_dirs":["any","any","any"]}, 
{"topic":"heima-par","partition":0,"replicas":[2,1,0],"log_dirs": ["any","any","any"]},
{"topic":"heima-par","partition":3,"replicas": [2,1,0],"log_dirs":["any","any","any"]}]} 

Proposed partition reassignment configuration 
{"version":1,"partitions":[{"topic":"heima-par","partition":0,"replicas": [1,2,3],"log_dirs":["any","any","any"]},
{"topic":"heima- par","partition":2,"replicas":[3,0,1],"log_dirs":["any","any","any"]}, 
{"topic":"heima-par","partition":1,"replicas":[2,3,0],"log_dirs": ["any","any","any"]},
{"topic":"heima-par","partition":3,"replicas": [0,1,2],"log_dirs":["any","any","any"]}]}

    上面命令中

    --generate 表示指定类型参数

    --topics-to-move-json-file 指定分区重分配对应的主题清单路径

    注意：
    命令输入两个Json字符串，第一个JSON内容为当前的分区副本分配情况，第二个为重新分配的候选方案，注意这里只是生成一份可行性的方案，并没有真正执行重分配的动作。

    我们将第二个JSON内容保存到名为result.json文件里面（文件名不重要，文件格式也不一定要以json为结尾，只要保证内容是json即可），然后执行这些reassign plan：

    重新分配JSON文件

    { 
    "version": 1, 
    "partitions": [ { 
        "topic": "heima-par", 
        "partition": 0, 
        "replicas": [ 1,2,3 ],
        "log_dirs": [ "any", "any", "any" ] 
    },{ 
        "topic": "heima-par", 
        "partition": 2, 
        "replicas": [ 3,0,1 ],
        "log_dirs": [ "any", "any", "any" ] 
    },{ 
        "topic": "heima-par", 
        "partition": 1, 
        "replicas": [ 2,3,0 ],
        "log_dirs": [ "any", "any", "any" ] 
    },{ 
        "topic": "heima-par", 
        "partition": 3, 
        "replicas": [ 0,1,2 ],
        "log_dirs": [ "any", "any", "any" ]
    } 
    ] 
}

    执行分配策略

    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ bin/kafka-reassign- 
partitions.sh --zookeeper localhost:2181 --reassignm 
ent-json-file result.json --execute 
Current partition replica assignment 

{"version":1,"partitions":[{"topic":"heima-par","partition":2,"replicas": [1,0,2],"log_dirs":["any","any","any"]},
{"topic":"heima- par","partition":1,"replicas":[0,2,1],"log_dirs":["any","any","any"]}, 
{"topic":"heima-par","partition":0,"replicas":[2,1,0],"log_dirs": ["any","any","any"]},
{"topic":"heima-par","partition":3,"replicas": [2,1,0],"log_dirs":["any","any","any"]}]} 
Save this to use as the --reassignment-json-file option during rollback 
Successfully started reassignment of partitions.

    查看分区重新分配的进度：

    itcast@Server-node:/mnt/d/kafka-cluster/kafka-1$ bin/kafka-reassign- 
partitions.sh --zookeeper localhost:2181 --reassignment-json-file result.json -- 
verify 
Status of partition reassignment: 
Reassignment of partition heima-par-3 completed successfully 
Reassignment of partition heima-par-0 is still in progress 
Reassignment of partition heima-par-2 is still in progress 
Reassignment of partition heima-par-1 is still in progress

    从上面信息可以看出 heima-par-3已经完成，其他三个正在进行中。

    四、修改副本因子

    场景

    实际项目中我们可能在创建topic时没有设置好正确的replication-factor，导致kafka集群虽然是高可用的，但是该topic在有broker宕机时，可能发生无法使用的情况。topic一旦使用又不能轻易删除重建，因此动态增加副本因子就成为最终的选择。

    说明：kafka 1.0版本配置文件默认没有 default.replication.factor=x， 因此如果创建topic时，不指定– replication-factor 想， 默认副本因子为1. 我们可以在自己的server.properties中配置上常用的副本因子，省去手动调整。例如设置 default.replication.factor=3

    首先我们配置topic的副本，保存为json文件

    { 
    "version":1, 
    "partitions":[ 
        {"topic":"heima","partition":0,"replicas":[0,1,2]}, 
        {"topic":"heima","partition":1,"replicas":[0,1,2]}, 
        {"topic":"heima","partition":2,"replicas":[0,1,2]} 
    ] 
}

    然后执行脚本 bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file replication-factor.json --execute

    itcast@Server-node:/mnt/d/kafka_2.12-2.2.1$ bin/kafka-reassign-partitions.sh -- 
zookeeper localhost:2181 --reassignment-json-file replication-factor.json -- 
execute 
Current partition replica assignment 

{"version":1,"partitions":[{"topic":"topic0703","partition":1,"replicas": [1,0],"log_dirs":["any","any"]},
{"topic":"topic0703","partition":0,"replicas": [0,1],"log_dirs":["any","any"]},
{"topic":"topic0703","partition":2,"replicas": [2,0],"log_dirs":["any","any"]}]} 

Save this to use as the --reassignment-json-file option during rollback 
Successfully started reassignment of partitions.

    验证

    itcast@Server-node:/mnt/d/kafka_2.12-2.2.1$ bin/kafka-topics.sh --describe -- 
zookeeper localhost:2181 --topic topic0703 
Topic:topic0703 PartitionCount:3    ReplicationFactor:3     Configs: 
Topic: topic0703    Partition: 0    Leader: 0   Replicas: 0,1,2 Isr: 0 
Topic: topic0703    Partition: 1    Leader: 1   Replicas: 0,1,2 Isr: 1,0 
Topic: topic0703    Partition: 2    Leader: 2   Replicas: 0,1,2 Isr: 2,0

    五、分区分配策略

    按照Kafka默认的消费逻辑设定，一个分区只能被同一个消费组（ConsumerGroup）内的一个消费者消费。假设目前某消费组内只有一个消费者C0，订阅了一个topic，这个topic包含7个分区，也就是说这个消费者C0订阅了7个分区，参考下图

    此时消费组内又加入了一个新的消费者C1，按照既定的逻辑需要将原来消费者C0的部分分区分配给消费者C1消费，情形上图（2），消费者C0和C1各自负责消费所分配到的分区，相互之间并无实质性的干扰。

    接着消费组内又加入了一个新的消费者C2，如此消费者C0、C1和C2按照上图（3）中的方式各自负责消费所分配到的分区。

    如果消费者过多，出现了消费者的数量大于分区的数量的情况，就会有消费者分配不到任何分区。参考下图，一共有8个消费者，7个分区，那么最后的消费者C7由于分配不到任何分区进而就无法消费任何消息。

    上面各个示例中的整套逻辑是按照Kafka中默认的分区分配策略来实施的。Kafka提供了消费者客户端参数
    partition.assignment.strategy用来设置消费者与订阅主题之间的分区分配策略。默认情况下，此参数的值为：org.apache.kafka.clients.consumer.RangeAssignor，即采用RangeAssignor分配策略。除此之外，Kafka中还提供了另外两种分配策略： RoundRobinAssignor和StickyAssignor。消费者客户端参数partition.asssignment.strategy可以配置多个分配策略，彼此之间以逗号分隔。

    1.RangeAssignor分配策略

    参考源码： org.apache.kafka.clients.consumer.RangeAssignor

    RangeAssignor策略 的原理是 按照消费者总数和分区总数进行整除运算来获得一个跨度，然后将分区按照跨度进行平均分配，以保证分区尽可能均匀地分配给所有的消费者。 对于每一个topic， RangeAssignor策略会将消费组内所有订阅这个topic的消费者按照名称的字典序排序，然后为每个消费者划分固定的分区范围，如果不够平均分配，那么字典序靠前的消费者会被多分配一个分区。

    假设n=分区数/消费者数量，m=分区数%消费者数量，那么前m个消费者每个分配n+1个分区，后面的（消费者数量-m）个消费者每个分配n个分区。

    假设消费组内有2个消费者C0和C1，都订阅了主题t0和t1，并且每个主题都有4个分区，那么所订阅的所有分区可以标识为：t0p0、t0p1、t0p2、t0p3、t1p0、t1p1、t1p2、t1p3。最终的分配结果为：

    消费者C0：t0p0、t0p1、t1p0、t1p1 
消费者C1：t0p2、t0p3、t1p2、t1p3

    假设上面例子中2个主题都只有3个分区，那么所订阅的所有分区可以标识为：t0p0、t0p1、t0p2、t1p0、t1p1、t1p2。最终的分配结果为：

    消费者C0：t0p0、t0p1、t1p0、t1p1 
消费者C1：t0p2、t1p2

    可以明显地看到这样的分配并不均匀，如果将类似的情形扩大，有可能会出现部分消费者过载的情况。

    2.RoundRobinAssignor分配策略

    参考源码： org.apache.kafka.clients.consumer.RoundRobinAssignor

    RoundRobinAssignor策略 的原理是将消费组内所有消费者以及消费者所订阅的所有topic的partition按照字典序排序，然后通过轮询方式逐个将分区以此分配给每个消费者。****RoundRobinAssignor策略对应的 partition.assignment.strategy参数值为： org.apache.kafka.clients.consumer.RoundRobinAssignor。

    假设消费组中有2个消费者C0和C1，都订阅了主题t0和t1，并且每个主题都有3个分区，那么所订阅的所有分区可以标识为：t0p0、t0p1、t0p2、t1p0、t1p1、t1p2。最终的分配结果为：

    消费者C0：t0p0、t0p2、t1p1 
消费者C1：t0p1、t1p0、t1p2

    如果同一个消费组内的消费者所订阅的信息是不相同的，那么在执行分区分配的时候就不是完全的轮询分配，有可能会导致分区分配的不均匀。如果某个消费者没有订阅消费组内的某个topic，那么在分配分区的时候此消费者将分配不到这个topic的任何分区。

    假设消费组内有3个消费者C0、C1和C2，它们共订阅了3个主题：t0、t1、t2，这3个主题分别有1、2、 3个分区，即整个消费组订阅了t0p0、t1p0、t1p1、t2p0、t2p1、t2p2这6个分区。具体而言，消费者C0订阅的是主题t0，消费者C1订阅的是主题t0和t1，消费者C2订阅的是主题t0、t1和t2，那么最终的分配结果为：

    消费者C0：t0p0 
消费者C1：t1p0 
消费者C2：t1p1、t2p0、t2p1、t2p2

    可以看到RoundRobinAssignor策略也不是十分完美，这样分配其实并不是最优解，因为完全可以将分区t1p1分配给消费者C1。

    3.StickyAssignor分配策略

    参考源码： org.apache.kafka.clients.consumer.StickyAssignor

    Kafka从0.11.x版本开始引入这种分配策略，它主要有两个目的：

    分区的分配要尽可能的均匀； 分区的分配尽可能的与上次分配的保持相同。

    当两者发生冲突时，第一个目标优先于第二个目标。鉴于这两个目标，StickyAssignor策略的具体实现要比RangeAssignor和RoundRobinAssignor这两种分配策略要复杂很多。

    假设消费组内有3个消费者：C0、C1和C2，它们都订阅了4个主题：t0、t1、t2、t3，并且每个主题有2个分区，也就是说整个消费组订阅了t0p0、t0p1、t1p0、t1p1、t2p0、t2p1、t3p0、t3p1这8个分区。最终的分配结果如下：

    消费者C0：t0p0、t1p1、t3p0 
消费者C1：t0p1、t2p0、t3p1 
消费者C2：t1p0、t2p1

    如分配结果所示，RoundRobinAssignor策略会按照消费者C0和C2进行重新轮询分配。而如果此时使用的是StickyAssignor策略，那么分配结果为：

    消费者C0：t0p0、t1p1、t3p0、t2p0 
消费者C2：t1p0、t2p1、t0p1、t3p1

    可以看到分配结果中保留了上一次分配中对于消费者C0和C2的所有分配结果，并将原来消费者C1的“负担”分配给了剩余的两个消费者C0和C2，最终C0和C2的分配还保持了均衡。

    4.自定义分配策略

    需实现： org.apache.kafka.clients.consumer.internals.PartitionAssignor

    继承自： org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor

    总结

    这篇讲解了对分区及副本的一系列操作，如分区副本机制、分区重新分配、修改副本因子等。

    相关文章

      网友评论

        本文标题：慕了，我要是早点看到这篇写 Kafka 的分区管理的文章就好了

        本文链接：https://www.haomeiwen.com/subject/qqncdltx.html

        延伸阅读

        深度阅读

        • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

        栏目导航

        热点阅读

        kafka

        关于我们|服务条款|联系我们|慕了，我要是早点看到这篇写 Kafka 的分区管理的文章就好了|投稿指南|网站地图|RSS订阅|排版工具|手机版

        提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

        Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

        备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

        本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

        所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！