Kafka源码分析-Consumer（4）-SubscriptionState分析

KafkaConsumer从Kafka拉取消息时发送的请求时FetchRequest,其中需要指定消费者希望拉取的起始消息的offset。为了消费者快速获取这个值，KafkaConsumer使用SubscriptionState来追踪TopicPartition和offset对应关系。

image.png

SubscriptionType是SubscriptionState的一个内部枚举类型，表示订阅Topic的模式：

NONE: SubscriptionState.subscriptionType的初始值。
AUTO_TOPICS: 按照指定的Topic名字进行订阅，自动分配分区。
AUTO_PARTITION: 按照指定的正则表达式匹配Topic进行订阅，自动分配分区。
AUTO_ASSIGNED: 用户手动指定消费者消费的Topic以及分区编号。
TopicPartitionState表示TopicPartition的消费状态，下面是相关的field介绍：
position:记录下次要从Kafka服务端获取的消息的offset。
committed:记录了最近一次提交的offset。
paused:记录了当前TopicPartition是否处于暂停状态，与Consumer接口的pause()方法相关。
resetStrategy:OffsetResetStrategy枚举类型，重置position的策略。同时，此字段是否为空，也表示了是否需要重置position的值。
同时TopicPartitionState提供了管理上面四个字段的方法。
介绍Comsumer接口时说过subscribe()方法和assign()方法是互斥的。其实上面介绍的三种消费模式都是都是互斥的。下面是setSubscriptionType()方法的代码,那三种订阅模式都会调用：

image.png

/**
     * This method sets the subscription type if it is not already set (i.e. when it is NONE),
     * or verifies that the subscription type is equal to the give type when it is set (i.e.
     * when it is not NONE)
     * @param type The given subscription type
     */
    private void setSubscriptionType(SubscriptionType type) {
        //如果是NONE，则可以指定其他模式
        if (this.subscriptionType == SubscriptionType.NONE)
            this.subscriptionType = type;
        else if (this.subscriptionType != type)//如果已经指定了其他模式，就会报错
            throw new IllegalStateException(SUBSCRIPTION_EXCEPTION_MESSAGE);
    }

SubscriptionState核心字段：

subscriptionType：表示订阅模式。
subscriptionPattern:使用AUTO_TOPICS模式时，对符合正在表达式的Topic进行订阅。
subscription：如果使用AUTO_TOPICS或AUTO_PARTITION模式，则使用此集合记录所有订阅的Topic。向subscription集合中添加数据的方法有两处：

changeSubscription的调用.jpg

在上图中的2处，使用的是AUTO_TOPICS模式订阅；1处是使用AUTO_PARTITION模式订阅。前面分析生产者的时候介绍过Metadata,可以在Metadata上添加listener，当Metadata发生变化时会触发
Metadata.Listener.onMetadataUpdate()方法，1处就是在Metadata的Listener中通过subscribedPattern模式过滤Topic，并调用changeSubscription()方法修改subscription集合。
userAssignment:如果使用AUTO_ASSIGNED模式，这个集合记录了分配给当前消费者的TopicPartition集合。SubscriptionType的模式是互斥的，所以对应的集合userAssignment和subscription集合也是互斥的。
assignment: Map<TopicPartition,TopicPartitionState>类型，无论用什么订阅模式，都用这个集合保存每个TopicPartition的消费状态。
groupSubscription:在前面描述的协议中，Consumer Group中会选一个Leader,Leader会使用这个集合记录Consumer Group中所有消费者订阅的Topic,而其他的Follower的这个集合只会保存自身订阅的Topic。
groupSubscription集合相关的用法：

SubscriptionState集合的使用 (1).png

1处是将消费者自身订阅的Topic添加到groupSubscription集合；2处是在Leader收到JoinGroupResponse时调用，在JoinGroupResponse中包含了全部消费者订阅的Topic，在此将Topic信息添加到groupSubscription集合。3处是将groupSubscription中删除其他消费者订阅的Topic,只留下本身订阅的topic（即subscription集合）。
needsPartitionAssignment:标记是否需要进行一次分区分配。下面的几种情况会对needsPartitionAssignment修改：

needsPartitionAssignment的调用 (1).jpg

上图中1，5处将needsPartitionAssignment设置为true是因为消费者订阅的Topic发生了变化，所以需要进行分区分配;3处将needsPartitionAssignment设置为false是因为使用了AUTO_ASSIGNED订阅模式，不需要分配分配操作；4是成功得到SyncGroupResponse中的分区分配结果时的操作，此时Rebalance操作结束，将needsPartitionAssignment设置为false；2处的场景比较复杂，调用2处将needsPartitionAssignment设置为true，主要是因为在某些请求响应中出现了ILLEGAL_GENERATION等异常，或是订阅的Topic出现了分区数量的变化，
needsFetchCommittedOffsets：标记了是否从GroupCoordinator获取最新提交的offset。当出现异步提交offset或Rebalance操作刚完成时会将其设置为true,成功获取最近提交的offset后会设置为false。
defaultResetStrategy:默认OffsetResetStrategy策略。
listener: ConsumerRebalanceListener类型，用于监听分区分配操作。
SubscriptionState中的方法主要是管理上面的字段，下面是subscribe()方法的分析：

public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
        
        //用户未指定ConsumerRebalanceListener时，默认使用NoOpConsumerRebalanceListener，但是
        //所有的方法都是空的。
        if (listener == null)
            throw new IllegalArgumentException("RebalanceListener cannot be null");
      setSubscriptionType(SubscriptionType.AUTO_TOPICS);//选择AUTO_TOPICS模式

        this.listener = listener;

        changeSubscription(topics);
    }


public void changeSubscription(Collection<String> topicsToSubscribe) {
        //如果订阅的Topic发生了变化
        if (!this.subscription.equals(new HashSet<>(topicsToSubscribe))) {
            this.subscription.clear();//情况subscription集合
            this.subscription.addAll(topicsToSubscribe);//添加订阅的Topic
            this.groupSubscription.addAll(topicsToSubscribe);
            this.needsPartitionAssignment = true;//标记需要重新分配分区

            // Remove any assigned partitions which are no longer subscribed to
            for (Iterator<TopicPartition> it = assignment.keySet().iterator(); it.hasNext(); ) {
                TopicPartition tp = it.next();
                if (!subscription.contains(tp.topic()))
                    it.remove();
            }
        }
    }