KafkaConsumer从Kafka拉取消息时发送的请求时FetchRequest,其中需要指定消费者希望拉取的起始消息的offset。为了消费者快速获取这个值，KafkaConsumer使用SubscriptionState来追踪TopicPartition和offset对应关系。
image.png
image.png
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();
}
}
}
网友评论