kafka中的消息分区分配算法怎么用

魁首哥

作者

kafka中的消息分区分配算法怎么用

本文小编为大家详细介绍“kafka中的消息分区分配算法怎么用”，内容详细，步骤清晰，细节处理妥当，希望这篇“kafka中的消息分区分配算法怎么用”文章能帮助大家解决疑惑，下面跟着小编的思路慢慢深入，一起来学习新知识吧。

背景

kafka有分区机制，一个主题topic在创建的时候，会设置分区。如果只有一个分区，那所有的消费者都订阅的是这一个分区消息；如果有多个分区的话，那消费者之间又是如何分配的呢？

分配算法

RangeAssignor

定义

Kafka默认采⽤RangeAssignor的分配算法。

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

这种分配⽅式明显的⼀个问题是随着消费者订阅的Topic的数量的增加，不均衡的问题会越来越严重，⽐如上图中4个分区3个消费者的场景，C0会多分配⼀个分区。如果此时再订阅⼀个分区数为4的Topic，那么C0⼜会⽐C1、C2多分配⼀个分区，这样C0总共就⽐C1、C2多分配两个分区了，⽽且随着Topic的增加，这个情况会越来越严重。

源码分析

publicclassRangeAssignorextendsAbstractPartitionAssignor{....@OverridepublicMap>assign(MappartitionsPerTopic,Mapsubscriptions){//1.获取每个topic被多少个consumer订阅了Map<String,List<String>>consumersPerTopic=consumersPerTopic(subscriptions);//2.存储最终的分配⽅案Map<String,List<String>>assignment=newHashMap<>();for(StringmemberId:subscriptions.keySet())assignment.put(memberId,newArrayList());for(Map.Entry>topicEntry:consumersPerTopic.entrySet()){Stringtopic=topicEntry.getKey();ListconsumersForTopic=topicEntry.getValue();//3.每个topic的partition数量IntegernumPartitionsForTopic=partitionsPerTopic.get(topic);if(numPartitionsForTopic==null)continue;Collections.sort(consumersForTopic);//4.表示平均每个consumer会分配到多少个partitionintnumPartitionsPerConsumer=numPartitionsForTopic/consumersForTopic.size();//5.平均分配后还剩下多少个partition未被分配intconsumersWithExtraPartition=numPartitionsForTopic%consumersForTopic.size();Listpartitions=AbstractPartitionAssignor.partitions(topic,numPartitionsForTopic);//6.这⾥是关键点,分配原则是将未能被平均分配的partition分配到前consumersWithExtraPartition个consumerfor(inti=0,n=consumersForTopic.size();i<n;i++){intstart=numPartitionsPerConsumer*i+Math.min(i,consumersWithExtraPartition);intlength=numPartitionsPerConsumer+(i+1>consumersWithExtraPartition?0:1);assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start,start+length));}}returnassignment;}}

场景

可以完全平均分配

无法完全平均分配，排序靠前分的更多

消费者数量大于分区数量，排名靠前先分得，排名靠后未分得分区

RoundRobinAssignor

定义

RoundRobinAssignor的分配策略是将消费组内订阅的所有Topic的分区及所有消费者进⾏排序后尽量均衡的分配（RangeAssignor是针对单个Topic的分区进⾏排序分配的）。如果消费组内，消费者订阅的Topic列表是相同的（每个消费者都订阅了相同的Topic），那么分配结果是尽量均衡的（消费者之间分配到的分区数的差值不会超过1）。

源码分析

packageorg.apache.kafka.clients.consumer;publicclassRoundRobinAssignorextendsAbstractPartitionAssignor{@OverridepublicMap>assign(MappartitionsPerTopic,Mapsubscriptions){<Map>assignment=newHashMap<>();for(StringmemberId:subscriptions.keySet())assignment.put(memberId,newArrayList());//1.环状链表,存储所有的consumer,⼀次迭代完之后⼜会回到原点CircularIteratorassigner=newCircularIterator<>(Utils.sorted(subscriptions.keySet()));//2.获取所有订阅的topic的partition总数for(TopicPartitionpartition:allPartitionsSorted(partitionsPerTopic,subscriptions)){finalStringtopic=partition.topic();while(!subscriptions.get(assigner.peek()).topics().contains(topic))assigner.next();assignment.get(assigner.next()).add(partition);}returnassignment;}....}

场景

无法完全平均分配，排序靠前分的更多

StickyAssignor

定义

尽管RoundRobinAssignor已经在RangeAssignor上做了⼀些优化来更均衡的分配分区，但是在⼀些情况下依旧会产⽣严重的分配偏差，从字⾯意义上看，Sticky是“粘性的”，可以理解为分配结果是带“粘性的”——每⼀次分配变更相对上⼀次分配做最少的变动（上⼀次的结果是有粘性的）其⽬标有两点：