public class RangeAssignor extends AbstractPartitionAssignor {
    ....
    @Override 
    public Map> assign(Map partitionsPerTopic, Map subscriptions) { 
        // 1. 獲取每個(gè)topic被多少個(gè)consumer訂閱了 
        Map<String,List<String>> consumersPerTopic = consumersPerTopic(subscriptions); 
        // 2. 存儲(chǔ)最終的分配?案 
        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. 每個(gè)topic的partition數(shù)量 
            Integer numPartitionsForTopic = partitionsPerTopic.get(topic); 
            if (numPartitionsForTopic == null) 
            continue; 
            Collections.sort(consumersForTopic); 
            // 4. 表示平均每個(gè)consumer會(huì)分配到多少個(gè)partition 
            int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size(); 
            // 5. 平均分配后還剩下多少個(gè)partition未被分配 
            int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size(); 
            List partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic); 
            // 6. 這?是關(guān)鍵點(diǎn),分配原則是將未能被平均分配的partition分配到前 consumersWithExtraPartition個(gè)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. 環(huán)狀鏈表,存儲(chǔ)所有的consumer,?次迭代完之后?會(huì)回到原點(diǎn) 
        CircularIterator assigner = new CircularIterator<> (Utils.sorted(subscriptions.keySet())); // 2. 獲取所有訂閱的topic的partition總數(shù) 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; 
    } 
.... }