消费者——分区分配策略 - 969251639/study GitHub Wiki

class Subscription {
        private final List<String> topics;//消费者订阅的主题列表
        private final ByteBuffer userData;//用户缓存数据，序列化二进制字节码时用

        public Subscription(List<String> topics, ByteBuffer userData) {
            this.topics = topics;
            this.userData = userData;
        }

        public Subscription(List<String> topics) {
            this(topics, ByteBuffer.wrap(new byte[0]));
        }

        public List<String> topics() {
            return topics;
        }

        public ByteBuffer userData() {
            return userData;
        }

        @Override
        public String toString() {
            return "Subscription(" +
                    "topics=" + topics +
                    ')';
        }
    }

    class Assignment {
        private final List<TopicPartition> partitions;//消费者分配到的分区列表
        private final ByteBuffer userData;//用户缓存数据，序列化二进制字节码时用

        public Assignment(List<TopicPartition> partitions, ByteBuffer userData) {
            this.partitions = partitions;
            this.userData = userData;
        }

        public Assignment(List<TopicPartition> partitions) {
            this(partitions, ByteBuffer.wrap(new byte[0]));
        }

        public List<TopicPartition> partitions() {
            return partitions;
        }

        public ByteBuffer userData() {
            return userData;
        }

        @Override
        public String toString() {
            return "Assignment(" +
                    "partitions=" + partitions +
                    ')';
        }
    }

public abstract class AbstractPartitionAssignor implements PartitionAssignor {
    private static final Logger log = LoggerFactory.getLogger(AbstractPartitionAssignor.class);

    /**
     * Perform the group assignment given the partition counts and member subscriptions
     * @param partitionsPerTopic The number of partitions for each subscribed topic. Topics not in metadata will be excluded
     *                           from this map.
     * @param subscriptions Map from the memberId to their respective topic subscription
     * @return Map from each member to the list of partitions assigned to them.
     */
    //抽象方法，由具体子类实现
    public abstract Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                             Map<String, Subscription> subscriptions);

    @Override
    public Subscription subscription(Set<String> topics) {//返回订阅的载体Subscription
        return new Subscription(new ArrayList<>(topics));
    }

    @Override
    public Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions) {//分配分区
        Set<String> allSubscribedTopics = new HashSet<>();//记录该消费者下所有的主题
        for (Map.Entry<String, Subscription> subscriptionEntry : subscriptions.entrySet())
            allSubscribedTopics.addAll(subscriptionEntry.getValue().topics());

        Map<String, Integer> partitionsPerTopic = new HashMap<>();
        for (String topic : allSubscribedTopics) {//遍历主题
            Integer numPartitions = metadata.partitionCountForTopic(topic);//获取主题下的分区数
            if (numPartitions != null && numPartitions > 0)//如果该主题已有分区
                partitionsPerTopic.put(topic, numPartitions);//缓存到partitionsPerTopic 
            else
                log.debug("Skipping assignment for topic {} since no metadata is available", topic);
        }

        Map<String, List<TopicPartition>> rawAssignments = assign(partitionsPerTopic, subscriptions);//根据不同的分区分配算法，返回每个主题下的分配到的分区

        // this class maintains no user data, so just wrap the results
        //处理分配结果后返回
        Map<String, Assignment> assignments = new HashMap<>();
        for (Map.Entry<String, List<TopicPartition>> assignmentEntry : rawAssignments.entrySet())
            assignments.put(assignmentEntry.getKey(), new Assignment(assignmentEntry.getValue()));
        return assignments;
    }

    @Override
    public void onAssignment(Assignment assignment) {//消费者收到leader分配结果时的毁掉方法，这里是空实现，一般由子类覆盖
        // this assignor maintains no internal state, so nothing to do
    }

    protected static <K, V> void put(Map<K, List<V>> map, K key, V value) {
        List<V> list = map.computeIfAbsent(key, k -> new ArrayList<>());
        list.add(value);
    }

    protected static List<TopicPartition> partitions(String topic, int numPartitions) {
        List<TopicPartition> partitions = new ArrayList<>(numPartitions);
        for (int i = 0; i < numPartitions; i++)
            partitions.add(new TopicPartition(topic, i));
        return partitions;
    }
}

.define(PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
    Type.LIST,
    Collections.singletonList(RangeAssignor.class),
    new ConfigDef.NonNullValidator(),
    Importance.MEDIUM,
    PARTITION_ASSIGNMENT_STRATEGY_DOC)