详解kafka中的消息分区分配算法

public class RangeAssignor extends AbstractPartitionAssignor {
    ....
    @Override 
    public Map> assign(Map partitionsPerTopic, Map subscriptions) { 
        // 1. 获取每个topic被多少个consumer订阅了 
        Map<String,List<String>> consumersPerTopic = consumersPerTopic(subscriptions); 
        // 2. 存储最终的分配⽅案 
        Map<String,List<String>> assignment = new HashMap<>(); 
        for (String memberId : subscriptions.keySet()) 
            assignment.put(memberId, new ArrayList()); 
        for (Map.Entry> topicEntry : consumersPerTopic.entrySet()) { 
            String topic = topicEntry.getKey(); 
            List consumersForTopic = topicEntry.getValue(); 
            // 3. 每个topic的partition数量 
            Integer numPartitionsForTopic = partitionsPerTopic.get(topic); 
            if (numPartitionsForTopic == null) 
            continue; 
            Collections.sort(consumersForTopic); 
            // 4. 表示平均每个consumer会分配到多少个partition 
            int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size(); 
            // 5. 平均分配后还剩下多少个partition未被分配 
            int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size(); 
            List partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic); 
            // 6. 这⾥是关键点,分配原则是将未能被平均分配的partition分配到前 consumersWithExtraPartition个consumer
            for (int i = 0, n = consumersForTopic.size(); i < n; i++) { 
                int start = numPartitionsPerConsumer * i + Math.min(i, consumersWithExtraPartition); 
                int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1); assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start, start + length)); 
                } 
            } 
            return assignment; 
    }
}

package org.apache.kafka.clients.consumer; 
public class RoundRobinAssignor extends AbstractPartitionAssignor { 
@Override 
public Map> assign(Map partitionsPerTopic, Map subscriptions) { 
        <Map> assignment = new HashMap<>(); 
        for (String memberId : subscriptions.keySet()) assignment.put(memberId, new ArrayList()); // 1. 环状链表,存储所有的consumer,⼀次迭代完之后⼜会回到原点 
        CircularIterator assigner = new CircularIterator<> (Utils.sorted(subscriptions.keySet())); // 2. 获取所有订阅的topic的partition总数 for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) { 
        final String topic = partition.topic(); 
        while (!subscriptions.get(assigner.peek()).topics().contains(topic)) 
            assigner.next(); 
            assignment.get(assigner.next()).add(partition); 
        }
        return assignment; 
    } 
.... }