Kafka与Storm的集成

//发送的第一步是添加到需要发送的列表中。

protected void setWaitingToEmit(ConsumerRecordsK, V ConsumerRecords) {

for (TopicPartition tp :consumerRecords.partitions()) {

waitingToEmit.put(tp, new ArrayList(consumerRecords.records(tp)));

}

}

受保护的无效emitIfWaitingNotEmited（）{

//从waitingToEmit列表中获取要发送的事件并开始发送

IteratorListConsumerRecordK, V waitingToEmitIter=waitingToEmit.values().iterator();

LOG.info('真实事件发出开始');

外环：

while (waitingToEmitIter.hasNext()) {

ListConsumerRecordK, V waitingToEmitForTp=waitingToEmitIter.next();

while (!waitingToEmitForTp.isEmpty()) {

最终布尔值已发射Tuple=emitOrRetryTuple(waitingToEmitForTp.remove(0));

如果（发射元组）{

LOG.error('事件发出失败');

打破外循环；

}

}

waitingToEmitIter.remove();

}

}

//实际发送过程

私人布尔emitOrRetryTuple（ConsumerRecordK，V记录）{

最终TopicPartition tp=new TopicPartition(record.topic(), record.partition());

最终KafkaSpoutMessageId msgId=retryService.getMessageId(record);

if (offsetManagers.containsKey(tp) offsetManagers.get(tp).contains(msgId)) { //已被确认

LOG.trace('记录[{}] 的元组已被确认。正在跳过', record);

} else if (emissed.contains(msgId)) { //已发出并且正在等待确认或失败

LOG.trace('记录[{}] 的元组已发出。正在跳过', record);

} 别的{

最终OffsetAndMetadata commitOffset=kafkaConsumer.commissed(tp);

如果（isAtLeastOnceProcessing（）

已提交偏移！=null

commitOffset.offset() record.offset()

commitMetadataManager.isOffsetCommitedByThisTopology（tp，commitedOffset，Collections.unmodifyingMap（offsetManagers）））{

//确保在启动具有此id 的拓扑后，消费者会获取

//位置永远不会落后于提交的偏移量(STORM-2844)

throw new IllegalStateException('尝试发出已提交的消息。'

+ '当使用至少一次处理保证时，永远不会发生这种情况。');

}

最终ListObject元组=kafkaSpoutConfig.getTranslator().apply(record);

如果（isEmitTuple（元组））{

最终布尔isScheduled=retryService.isScheduled(msgId);

//未调度=从未失败（即从未发出），或已调度并准备好重试

if (!isScheduled || retryService.isReady(msgId)) {

最终字符串流=KafkaTuple 的元组实例？ ((KafkaTuple)元组).getStream() : Utils.DEFAULT_STREAM_ID;

if (!isAtLeastOnceProcessing()) {

如果（kafkaSpoutConfig.isTupleTrackingEnforced（））{

收集器.emit（流，元组，msgId）；

LOG.trace('使用msgId [{}] 为记录[{}] 发出元组[{}]', tuple, record, msgId);

} 别的{

收集器.emit（流，元组）；

LOG.trace('为记录[{}] 发出元组[{}]', tuple, record);

}

} 别的{

发射。添加（msgId）；

offsetManagers.get(tp).addToEmitMsgs(msgId.offset());

if (isScheduled) { //已安排重试并重新发出，因此从计划中删除。

retryService.remove(msgId);

}

收集器.emit（流，元组，msgId）；

tupleListener.onEmit(tuple, msgId);

LOG.trace('使用msgId [{}] 为记录[{}] 发出元组[{}]', tuple, record, msgId);

}

返回真；

}

} 别的{

/*如果空元组未配置为发出，则应将其标记为已发出并立即确认

* 允许将其偏移量提交给Kafka*/

LOG.debug('不按照配置中的定义发出记录[{}] 的空元组。', record);

如果（isAtLeastOnceProcessing（））{

msgId.setNullTuple(true);

offsetManagers.get(tp).addToEmitMsgs(msgId.offset());

确认（消息ID）；

}

}

}

返回假；

}