一、属性
//org.apache.storm.spout.SpoutOutputCollector
protected SpoutOutputCollector collector;
//org.apache.storm.kafka.spout.KafkaSpoutConfig
private final KafkaSpoutConfig<K, V> kafkaSpoutConfig;
//consumer工厂
private KafkaConsumerFactory kafkaConsumerFactory;
//consumer
private transient KafkaConsumer<K, V> kafkaConsumer;
//是否是consume自动提交
private transient boolean consumerAutoCommitMode;
//首次拉取消息的offset
private transient FirstPollOffsetStrategy firstPollOffsetStrategy;
//重试机制
private transient KafkaSpoutRetryService retryService;
//提交offset的定时器
private transient Timer commitTimer;
//是否已经初始化
private transient boolean initialized;
//offset管理器
private transient Map<TopicPartition, OffsetManager> offsetManagers;
//已经发射的元组集合
private transient Set<KafkaSpoutMessageId> emitted;
//等待发射的元组
private transient Iterator<ConsumerRecord<K, V>> waitingToEmit;
//允许未提交的offset数量，达到这个值后将提交offset
private transient long numUncommittedOffsets;
//刷新订阅主题的定时器
private transient Timer refreshSubscriptionTimer;
//拓扑关系上下文
private transient TopologyContext context;
其中：
private transient Map<TopicPartition, OffsetManager> offsetManagers;
private transient Set<KafkaSpoutMessageId> emitted;
private transient Iterator<ConsumerRecord<K, V>> waitingToEmit;
KafkaSpout就是围绕着这3个集合进行的。
当拉取消息后首先放到waitingToEmit，然后发射后放到emitted中，最后ack后放到offsetManagers。理解了这3个的关系就很容易理解整个流程了
二、构造方法
KafkaSpout继承自BaseRichSpout并覆盖实现了（因为父类是空实现）所有方法，而BaseRichSpout又extends BaseComponent implements IRichSpout。因此KafkaSpout整体结构如下：
KafkaSpout的构造方法：
//传入一个KafkaSpoutConfig来构造KafkaSpout，使用默认的consumer factory
public KafkaSpout(KafkaSpoutConfig<K, V> kafkaSpoutConfig) {
this(kafkaSpoutConfig, new KafkaConsumerFactoryDefault());
}
//传入KakfaSpoutConfig和KafkaConsumerFactory来构造
KafkaSpout(KafkaSpoutConfig<K, V> kafkaSpoutConfig, KafkaConsumerFactory<K, V> kafkaConsumerFactory) {
this.kafkaConsumerFactory = kafkaConsumerFactory;
this.kafkaSpoutConfig = kafkaSpoutConfig;
}
其中KafkaSpoutConfig包含了对KafkaSpout的所有配置以及对应consumer的配置。
KafkaConsumerFactoryDefault()是一个默认的consumer factory实现，如下：
public class KafkaConsumerFactoryDefault<K, V> implements KafkaConsumerFactory<K, V> {
public KafkaConsumerFactoryDefault() {
}
public KafkaConsumer<K, V> createConsumer(KafkaSpoutConfig<K, V> kafkaSpoutConfig) {
return new KafkaConsumer(kafkaSpoutConfig.getKafkaProps(), kafkaSpoutConfig.getKeyDeserializer(), kafkaSpoutConfig.getValueDeserializer());
}
}
三、主要方法解析
1.open()
实现自ISpout，主要是进行一些初始化
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
//未初始化
this.initialized = false;
//设置上下文
this.context = context;
//收集器
this.collector = collector;
//允许未提交的Offset数量，默认0，及实时提交
this.numUncommittedOffsets = 0L;
//第一次拉取消息的offset，默认UNCOMMITTED_EARLIEST，即最早未提交
this.firstPollOffsetStrategy = this.kafkaSpoutConfig.getFirstPollOffsetStrategy();
//消费者的offset提交模式，默认false，即由Kafkaspout管理提交offset，若果为true则有consumer自动提交offset
this.consumerAutoCommitMode = this.kafkaSpoutConfig.isConsumerAutoCommitMode();
//重试服务
this.retryService = this.kafkaSpoutConfig.getRetryService();
//入股不是自动提交模式，则在定时提交，时间间隔由KafkaSpoutConfig传入
if (!this.consumerAutoCommitMode) {
this.commitTimer = new Timer(500L, this.kafkaSpoutConfig.getOffsetsCommitPeriodMs(), TimeUnit.MILLISECONDS);
}
//重新刷新订阅主题定时器，默认500ms，时间间隔由KafkaSpoutConfig传入
this.refreshSubscriptionTimer = new Timer(500L, this.kafkaSpoutConfig.getPartitionRefreshPeriodMs(), TimeUnit.MILLISECONDS);
//实例化
this.offsetManagers = new HashMap();
this.emitted = new HashSet();
this.waitingToEmit = Collections.emptyListIterator();
LOG.info("Kafka Spout opened with the following configuration: {}", this.kafkaSpoutConfig);
}
2.setAcked()
确认数据已被正确处理
当数据被bolt处理后，bolt调用ack后表明已被争取处理，在这里将TopicPartition和offset加入offsetManagers
private void setAcked(TopicPartition tp, long fetchOffset) {
if (!this.consumerAutoCommitMode && !this.offsetManagers.containsKey(tp)) {
this.offsetManagers.put(tp, new OffsetManager(tp, fetchOffset));
}
}
3.nextTuple()
发送元组。
大致逻辑如下：
首先判断是否已经完成初始化，如果是，则判断是否已经提交，如果已提交，则确认提交，接着判断是否允许从Kafka拉取数据，允许的条件是：
boolean poll = !this.waitingToEmit() && (this.numUncommittedOffsets - (long)readyMessageCount < (long)maxUncommittedOffsets || this.consumerAutoCommitMode);
即没有正在等待提交的并且未提交的数量-准备好的数量<最大允许未提交的数量，如果为true，则拉取消息并加入到等待提交的列表中；然后判断是否正在等待发射，如果是，则发射元组。看下边代码会更清楚一些。
public void nextTuple() {
try {
//如果已经初始化
if (this.initialized) {
//如果已经提交
if (this.commit()) {
//确认提交
this.commitOffsetsForAckedTuples();
}
//如果允许拉取
if (this.poll()) {
try {
//拉取Kafka数据并放进waitingToEmit集合中
this.setWaitingToEmit(this.pollKafkaBroker());
} catch (RetriableException var2) {
LOG.error("Failed to poll from kafka.", var2);
}
}
//如果正在等待发射，则发射
if (this.waitingToEmit()) {
this.emit();
}
} else {
LOG.debug("Spout not initialized. Not sending tuples until initialization completes");
}
} catch (InterruptException var3) {
this.throwKafkaConsumerInterruptedException();
}
}
5.poll()
private boolean poll() {
int maxUncommittedOffsets = this.kafkaSpoutConfig.getMaxUncommittedOffsets();
int readyMessageCount = this.retryService.readyMessageCount();
boolean poll = !this.waitingToEmit() && (this.numUncommittedOffsets - (long)readyMessageCount < (long)maxUncommittedOffsets || this.consumerAutoCommitMode);
if (!poll) {
if (this.waitingToEmit()) {
LOG.debug("Not polling. Tuples waiting to be emitted. [{}] uncommitted offsets across all topic partitions", this.numUncommittedOffsets);
}
if (this.numUncommittedOffsets >= (long)maxUncommittedOffsets && !this.consumerAutoCommitMode) {
LOG.debug("Not polling. [{}] uncommitted offsets across all topic partitions has reached the threshold of [{}]", this.numUncommittedOffsets, maxUncommittedOffsets);
}
}
return poll;
}
6.commit()
判断是改提交。
如果 非 自动提交模式 并且 提交时间间隔已到则返回true
private boolean commit() {
return !this.consumerAutoCommitMode && this.commitTimer.isExpiredResetOnTrue();
}
7.waitingToEmit()
是否正在等待发射.
如果 waitingToEmit不为空 则返回true，表明有正在等待发射的
private boolean waitingToEmit() {
return this.waitingToEmit != null && this.waitingToEmit.hasNext();
}
8.setWaitingToEmit()
设置准备发射的。
作用就是将消费到的数据添加到一个等待提交的列表waitingToEmit中
public void setWaitingToEmit(ConsumerRecords<K, V> consumerRecords) {
List<ConsumerRecord<K, V>> waitingToEmitList = new LinkedList();
Iterator var3 = consumerRecords.partitions().iterator();
//遍历所有主题分区
while(var3.hasNext()) {
//得到主题分区
TopicPartition tp = (TopicPartition)var3.next();
//添加对应的主题分区数据
waitingToEmitList.addAll(consumerRecords.records(tp));
}
//添加到等待发射的
this.waitingToEmit = waitingToEmitList.iterator();
}
9.pollKafkaBroker
拉取Kafka数据
private ConsumerRecords<K, V> pollKafkaBroker() {
//先处理未成功的进行重置offset操作
this.doSeekRetriableTopicPartitions();
//如果到了刷新订阅时间，则先刷新
if (this.refreshSubscriptionTimer.isExpiredResetOnTrue()) {
this.kafkaSpoutConfig.getSubscription().refreshAssignment();
}
//拉取数据
ConsumerRecords<K, V> consumerRecords = this.kafkaConsumer.poll(this.kafkaSpoutConfig.getPollTimeoutMs());
int numPolledRecords = consumerRecords.count();
LOG.debug("Polled [{}] records from Kafka. [{}] uncommitted offsets across all topic partitions", numPolledRecords, this.numUncommittedOffsets);
return consumerRecords;
}
10.doSeekRetriableTopicPartitions
private void doSeekRetriableTopicPartitions() {
//先获取需要重试的集合
Map<TopicPartition, Long> retriableTopicPartitions = this.retryService.earliestRetriableOffsets();
Iterator var2 = retriableTopicPartitions.entrySet().iterator();
while(var2.hasNext()) {
Entry<TopicPartition, Long> retriableTopicPartitionAndOffset = (Entry)var2.next();
//重新定位offset
this.kafkaConsumer.seek((TopicPartition)retriableTopicPartitionAndOffset.getKey(), ((Long)retriableTopicPartitionAndOffset.getValue()).longValue());
}
}
11.emit（）
发射元组。注意：这里只是对等待发射的waitingToEmit列表进行操作。
private void emit() {
//如果没有发射并且存在等待发射的，则将正在等待的从列表中移除，然后进行发射（发射操作在nextTuple()中）
while(!this.emitTupleIfNotEmitted((ConsumerRecord)this.waitingToEmit.next()) && this.waitingToEmit.hasNext()) {
this.waitingToEmit.remove();
}
}
12.emitTupleIfNotEmitted（）
发射没有发射的元组
主要逻辑如下：
首先判断原则是否已经被确认，
然后判断是否已经发射，
满足以上两个条件之一时说明不需要发射。
然后判断需不需要发射tuple：主要是判断tuple是否为空，以及配置中是否允许发射空的tuple。如果需要发射则：
先判断重试列表中有没有当前tuple，若有，则判断是否需要重发，若需要，则继续判断是否是自动提交offset模式，若自动提交则直接发射，否则加入offsetManagers，然后判断是否在重试列表中，是则从重试列表中移除，否则将未提交offset数量加1，然后发射
private boolean emitTupleIfNotEmitted(ConsumerRecord<K, V> record) {
TopicPartition tp = new TopicPartition(record.topic(), record.partition());
KafkaSpoutMessageId msgId = this.retryService.getMessageId(record);
if (this.offsetManagers.containsKey(tp) && ((OffsetManager)this.offsetManagers.get(tp)).contains(msgId)) {
LOG.trace("Tuple for record [{}] has already been acked. Skipping", record);
} else if (this.emitted.contains(msgId)) {
LOG.trace("Tuple for record [{}] has already been emitted. Skipping", record);
} else {
List<Object> tuple = this.kafkaSpoutConfig.getTranslator().apply(record);
if (this.isEmitTuple(tuple)) {
boolean isScheduled = this.retryService.isScheduled(msgId);
if (!isScheduled || this.retryService.isReady(msgId)) {
if (this.consumerAutoCommitMode) {
if (tuple instanceof KafkaTuple) {
this.collector.emit(((KafkaTuple)tuple).getStream(), tuple);
} else {
this.collector.emit(tuple);
}
} else {
this.emitted.add(msgId);
((OffsetManager)this.offsetManagers.get(tp)).addToEmitMsgs(msgId.offset());
if (isScheduled) {
this.retryService.remove(msgId);
} else {
++this.numUncommittedOffsets;
}
if (tuple instanceof KafkaTuple) {
this.collector.emit(((KafkaTuple)tuple).getStream(), tuple, msgId);
} else {
this.collector.emit(tuple, msgId);
}
}
LOG.trace("Emitted tuple [{}] for record [{}] with msgId [{}]", new Object[]{tuple, record, msgId});
return true;
}
} else {
LOG.debug("Not emitting null tuple for record [{}] as defined in configuration.", record);
msgId.setEmitted(false);
this.ack(msgId);
}
}
return false;
}
12.isEmitTuple（）
是否发送元组。先判空，然后在判断config中是否允许发射空原则，满足一个即可
private boolean isEmitTuple(List<Object> tuple) {
return tuple != null || this.kafkaSpoutConfig.isEmitNullTuples();
}
13.commitOffsetsForAckedTuples（）
提交已经确认的Tuple的偏移量
private void commitOffsetsForAckedTuples() {
//用来保存将要提交的数据
Map<TopicPartition, OffsetAndMetadata> nextCommitOffsets = new HashMap();
//从offsetManagers获取数据
Iterator var2 = this.offsetManagers.entrySet().iterator();
Entry tpOffset;
while(var2.hasNext()) {
tpOffset = (Entry)var2.next();
OffsetAndMetadata nextCommitOffset = ((OffsetManager)tpOffset.getValue()).findNextCommitOffset();
if (nextCommitOffset != null) {
nextCommitOffsets.put(tpOffset.getKey(), nextCommitOffset);
}
}
if (!nextCommitOffsets.isEmpty()) {
//进行提交操作
this.kafkaConsumer.commitSync(nextCommitOffsets);
LOG.debug("Offsets successfully committed to Kafka [{}]", nextCommitOffsets);
var2 = nextCommitOffsets.entrySet().iterator();
//日志输出未提交的
while(var2.hasNext()) {
tpOffset = (Entry)var2.next();
TopicPartition tp = (TopicPartition)tpOffset.getKey();
OffsetManager offsetManager = (OffsetManager)this.offsetManagers.get(tp);
long numCommittedOffsets = offsetManager.commit((OffsetAndMetadata)tpOffset.getValue());
this.numUncommittedOffsets -= numCommittedOffsets;
LOG.debug("[{}] uncommitted offsets across all topic partitions", this.numUncommittedOffsets);
}
} else {
LOG.trace("No offsets to commit. {}", this);
}
}
14.ack（）
public void ack(Object messageId) {
//转化为KafkaMessageId
KafkaSpoutMessageId msgId = (KafkaSpoutMessageId)messageId;
//如果已经提交的不包含msgId
if (!this.emitted.contains(msgId)) {
//如果已经提交
if (msgId.isEmitted()) {
LOG.debug("Received ack for message [{}], associated with tuple emitted for a ConsumerRecord that came from a topic-partition that this consumer group instance is no longer tracking due to rebalance/partition reassignment. No action taken.", msgId);
} else {
LOG.debug("Received direct ack for message [{}], associated with null tuple", msgId);
}
} else {//如果不已提交列表不包含msgId
//如果非自动提交模式
if (!this.consumerAutoCommitMode) {
//添加到确认列表
((OffsetManager)this.offsetManagers.get(msgId.getTopicPartition())).addToAckMsgs(msgId);
}
//移除
this.emitted.remove(msgId);
}
}
15.fail()
失败处理。
主要操作就是增加失败次数，然后调用重试服务进行重试
public void fail(Object messageId) {
KafkaSpoutMessageId msgId = (KafkaSpoutMessageId)messageId;
if (!this.emitted.contains(msgId)) {
LOG.debug("Received fail for tuple this spout is no longer tracking. Partitions may have been reassigned. Ignoring message [{}]", msgId);
} else {
msgId.incrementNumFails();
if (!this.retryService.schedule(msgId)) {
LOG.debug("Reached maximum number of retries. Message [{}] being marked as acked.", msgId);
this.ack(msgId);
} else {
this.emitted.remove(msgId);
}
}
}