Kafka分区分配策略(Partition Assignment Strategy)
众所周知,Apache Kafka是基于生产者和消费者模型作为开源的分布式发布订阅消息系统(当然,目前Kafka定位于an open-source distributed event streaming platform),由Scala和Java编写。
Kafka提供了类似于JMS的特性,但设计上又有很大区别,它不是JMS规范的实现,如Kafka允许多个消费者主动拉取数据,而在JMS中只有点对点模式消费者才会主动拉取数据。
Kafka producer在向Kafka集群发送消息时,需要指定topic,Kafka根据topic对消息进行归类(逻辑划分),而一个topic通常会有多个partition分区,落到磁盘上就是多个partition目录。
Kafka consumer为了及时消费消息,会以Consumer Group(消费组)的形式,启动多个consumer消费消息。不同的消费组在消费消息时彼此互不影响,同一个消费组的consumer协调在一起消费订阅的topic所有分区消息。这就引申一个问题:消费组中的consumer是如何确定自己该消费哪些分区的数据的?
Kafka提供了多种分区策略如RoundRobin(轮询)、Range(按范围),可通过参数partition.assignment.strategy进行配置。
一般情况下,在topic和消费组不发生变化,Kafka会根据topic分区、消费组情况等确定分区策略,但是当发生以下情况时,会触发Kafka的分区重分配:
1. Consumer Group中的consumer发生了新增或者减少
- 同一个Consumer Group新增consumer
- Consumer Group订阅的topic分区发生变化如新增分区
2. Consumer Group订阅的topic分区发生变化如新增分区
本文通过下面的场景,来分别阐述Kafka主要的分配策略RoundRobin和Range:
Range Strategy
Range策略是针对topic而言的,在进行分区分配时,为了尽可能保证所有consumer均匀的消费分区,会对同一个topic中的partition按照序号排序,并对consumer按照字典顺序排序。
然后为每个consumer划分固定的分区范围,如果不够平均分配,那么排序靠前的消费者会被多分配分区。具体就是将partition的个数除于consumer线程数来决定每个consumer线程消费几个分区。如果除不尽,那么前面几个消费者线程将会多分配分区。
通过下面公式更直观:
假设n = 分区数 / 消费者数量,m = 分区数 % 消费者线程数量,那么前m个消费者每个分配n+1个分区,后面的(消费者线程数量 - m)个消费者每个分配n个分区。
举个例子:
一个消费组CG1中有C0和C1两个consumer,消费Kafka中的主题t1。t1的分区数为10,并且C1的num.streams为1,C2的num.streams为2。
经过排序后,分区为:0, 1, 2, 3, 4, 5, 6, 7, 8, 9;CG1中消费者线程为C0-0、C1-0、C1-1。然后因为 10除3除不尽,那么消费者线程C0-0将会多分配分区,所以分区分配之后结果如下:
C0-0 将消费0、1、2、3分区
C1-0 将消费4、5、6分区
C1-1 将消费7、8、9分区
当存在有2个Kafka topic(t1和t2),它们都有有10个partition,那么最后分区结果为:
C0-0 将消费t1主题的0、1、2、3分区以及t2主题的0、1、2、3分区
C1-0 将消费t1主题的4、5、6分区以及t2主题的4、5、6分区
C2-1 将消费t1主题的7、8、9分区以及t2主题的7、8、9分区
如上场景,随着topic的增多,那么针对每个topic,消费者C0-0都将多消费1个分区,topic越多比如为N个,C0-0消费的分区会比其他消费者明显多消费N个分区。
可以明显的看到这样的分配并不均匀,如果将类似的情形扩大,有可能会出现部分消费者过载的情况,这就是Range分区策略的一个很明显的弊端。
RoundRobin Strategy
RoundRobin策略的工作原理:将所有topic的partition组成TopicAndPartition列表,然后对TopicAndPartition列表按照hashCode进行排序:
val allTopicPartitions = ctx.partitionsForTopic.flatMap { case(topic, partitions) =>
info("Consumer %s rebalancing the following partitions for topic %s: %s"
.format(ctx.consumerId, topic, partitions))
partitions.map(partition => {
TopicAndPartition(topic, partition)
})
}.toSeq.sortWith((topicPartition1, topicPartition2) => {
/*
* Randomize the order by taking the hashcode to reduce the likelihood of all partitions of a given topic ending
* up on one consumer (if it has a high enough stream count).
*/
topicPartition1.toString.hashCode < topicPartition2.toString.hashCode
})
最后按照RoundRobin风格将分区分别分配给不同的消费者。
使用RoundRobin策略必须满足以下条件:
1.同一个Consumer Group里面的所有consumer的num.streams必须相等
2.每个consumer订阅的topic必须相同
假设消费组CG1中有C0和C1两个consumer的num.streams都为2。按照hashCode排序完的topic-partition组依次为t1-5, t1-3, t1-0, t1-8, t1-2, t1-1, t1-4, t1-7, t1-6, t1-9,我们的消费者排序为C0-0, C0-1, C1-0, C1-1,最后分区分配的结果为:
C0-0将消费t1-5、t1-2、t1-6分区
C0-1将消费t1-3、t1-1、t1-9分区
C1-0将消费t1-0、t1-4分区
C1-1将消费t1-8、t1-7分区
多个主题的分区分配和单个主题类似,这里就不在介绍了。
上面RoundRobin要求每个consumer订阅的topic必须相同,当订阅的topic不同时,那么在执行分区分配的时候就不是完全的轮询分配,有可能会导致分区分配的不均匀。比如,某个consumer没有订阅消费组内的某个topic,那么在分配分区的时候,这个consumer将分配不到这个topic的分区。
除了上述的介绍的RoundRobin和Range分配策略,Kafka还有Sticky分配策略,它主要有两个目的:
- 分区的分配要尽可能的均匀
- 分区的分配尽可能的与上次分配的保持相同
当两者发生冲突时,第一个目标优先于第二个目标。鉴于这两个目标,StickyAssignor策略的具体实现要比RangeAssignor和RoundRobinAssignor这两种分配策略要复杂很多。
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- keras输出预测值和真实值方式
- 对python pandas中 inplace 参数的理解
- PHP基于GD2函数库实现验证码功能示例
- PHP合并两个或多个数组的方法
- PHP的mysqli_set_charset()函数讲解
- PHP常见加密函数用法示例【crypt与md5】
- PHP的mysqli_sqlstate()函数讲解
- PHP Laravel中的Trait使用方法
- PHP CURL使用详解
- 解析Python 偏函数用法全方位实现
- PHP读取目录树的实现方法分析
- python suds访问webservice服务实现
- 使用Python-OpenCV消除图像中孤立的小区域操作
- PHP实现数组和对象的相互转换操作示例
- PHP单元测试框架PHPUnit用法详解