KafkaSpout源码注解

￡神魔★判官ぃ 2022-05-26 03:45 165阅读 0赞

# KafkaSpout源码注释（入门级） #

注：本文所有代码块合在一起即为KafkaSpout完整代码（持续更新）

### 模块引入： ###

package org.apache.storm.kafka; import com.google.common.base.Strings; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; import org.apache.storm.kafka.KafkaUtils.KafkaOffsetMetric; import org.apache.storm.kafka.PartitionManager.KafkaMessageId; import org.apache.storm.kafka.trident.GlobalPartitionInformation; import org.apache.storm.metric.api.IMetric; import org.apache.storm.spout.SpoutOutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.topology.base.BaseRichSpout; import org.slf4j.Logger; import org.slf4j.LoggerFactory;

### 变量定义 ###

public class KafkaSpout extends BaseRichSpout { private static final Logger LOG = LoggerFactory.getLogger(KafkaSpout.class); //指定类的初始化日志对象 SpoutConfig _spoutConfig; //Spout配置对象 SpoutOutputCollector _collector; //输出收集器（鬼名自己翻译的） PartitionCoordinator _coordinator; //分区协调器 DynamicPartitionConnections _connections; //动态分区连接 ZkState _state; //zookeeper连接 long _lastUpdateMs = 0L; //上次更新时间 int _currPartitionIndex = 0; //当前分区

### 构造函数 ###

public KafkaSpout(SpoutConfig spoutConf) { this._spoutConfig = spoutConf; //初始化spout配置 }

### Spout入口open函数 ###

public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { this._collector = collector; //初始化收集器 String topologyInstanceId = context.getStormId(); //获得拓扑名称 Map<String, Object> stateConf = new HashMap(conf); //用于保存zk的配置信息 List<String> zkServers = this._spoutConfig.zkServers; //以下获取zk信息，并存入stateConf if (zkServers == null) { zkServers = (List)conf.get("storm.zookeeper.servers"); }              Integer zkPort = this._spoutConfig.zkPort; if (zkPort == null) { zkPort = ((Number)conf.get("storm.zookeeper.port")).intValue(); } stateConf.put("transactional.zookeeper.servers", zkServers); stateConf.put("transactional.zookeeper.port", zkPort); stateConf.put("transactional.zookeeper.root", this._spoutConfig.zkRoot); this._state = new ZkState(stateConf); //初始化zk连接器         //初始化动态分区连接         this._connections = new DynamicPartitionConnections(this._spoutConfig, KafkaUtils.makeBrokerReader(conf, this._spoutConfig));                  //获得spout的数量，在TopologyBuilder.setSpout传入 int totalTasks = context.getComponentTasks(context.getThisComponentId()).size();          /**           * host用于获得ka去a的broker和partition的影射关系          * host有StaticHosts和ZkHosts两种实现方式           * StaticHosts：用于静态分区信息（需要传入GlobalPartitionInformation）           * ZkHosts：会动态的从zk中更新broker的信息（refreshFreqSecs=60)           * 以此为依据创建相应的协调器           * ZkCoordinator: 可以动态响应broker partition的变化动态刷新，重新分配manager          * StaticCoordinator: 静态      */ if (this._spoutConfig.hosts instanceof StaticHosts) { this._coordinator = new StaticCoordinator(this._connections, conf, this._spoutConfig, this._state, context.getThisTaskIndex(), totalTasks, topologyInstanceId); } else { this._coordinator = new ZkCoordinator(this._connections, conf, this._spoutConfig, this._state, context.getThisTaskIndex(), totalTasks, topologyInstanceId); }        //注册kafkaOffset信息 context.registerMetric("kafkaOffset", new IMetric() { KafkaOffsetMetric _kafkaOffsetMetric; { this._kafkaOffsetMetric = new KafkaOffsetMetric(KafkaSpout.this._connections); } public Object getValueAndReset() {                 // 获取当前task的对应的partitions List<PartitionManager> pms = KafkaSpout.this._coordinator.getMyManagedPartitions(); Set<Partition> latestPartitions = new HashSet(); Iterator var3 = pms.iterator(); PartitionManager pm; while(var3.hasNext()) { pm = (PartitionManager)var3.next(); latestPartitions.add(pm.getPartition()); } 
                         this._kafkaOffsetMetric.refreshPartitions(latestPartitions); var3 = pms.iterator(); while(var3.hasNext()) { pm = (PartitionManager)var3.next(); this._kafkaOffsetMetric.setOffsetData(pm.getPartition(), pm.getOffsetData()); } return this._kafkaOffsetMetric.getValueAndReset(); }
            }, this._spoutConfig.metricsTimeBucketSizeInSecs); context.registerMetric("kafkaPartition", new IMetric() { public Object getValueAndReset() { List<PartitionManager> pms = KafkaSpout.this._coordinator.getMyManagedPartitions(); Map concatMetricsDataMaps = new HashMap(); Iterator var3 = pms.iterator(); while(var3.hasNext()) { PartitionManager pm = (PartitionManager)var3.next(); concatMetricsDataMaps.putAll(pm.getMetricsDataMap()); } 
                  return concatMetricsDataMaps; } }, this._spoutConfig.metricsTimeBucketSizeInSecs); }

### 关闭连接 ###

public void close() { this._state.close(); }

###  ###

### tuple处理函数 ###

public void nextTuple() {         // 获得partition的managers，manager记录者当前task对某一partition的具体信息 List<PartitionManager> managers = this._coordinator.getMyManagedPartitions(); for(int i = 0; i < managers.size(); ++i) { try { this._currPartitionIndex %= managers.size();                 /**                 * 在next中会进行EMITED并返回操作之后的状态，状态信息如下：                 * NO_EMITED: 没有emit tuple                 * EMITED_MORE_LEFT: emit了tuple且有剩余，之后跟换manager（partition），跳出循环                 * EMITED_END: 完成了所有tuple的emit，跳出循环                 * 注：只要有一条消息被成功转为tuple循环就不在进行                 */ KafkaSpout.EmitState state = ((PartitionManager)managers.get(this._currPartitionIndex)).next(this._collector); if (state != KafkaSpout.EmitState.EMITTED_MORE_LEFT) { this._currPartitionIndex = (this._currPartitionIndex + 1) % managers.size(); } if (state != KafkaSpout.EmitState.NO_EMITTED) { break; } } catch (FailedFetchException var4) { LOG.warn("Fetch failed", var4); this._coordinator.refresh(); } }         // 根据spout的stateUpdateIntervalMs提交更新到zk long diffWithNow = System.currentTimeMillis() - this._lastUpdateMs; if (diffWithNow > this._spoutConfig.stateUpdateIntervalMs || diffWithNow < 0L) { this.commit(); } }

### 外部获取manager函数 ###

private PartitionManager getManagerForPartition(int partition) { Iterator var2 = this._coordinator.getMyManagedPartitions().iterator(); PartitionManager partitionManager; do { if (!var2.hasNext()) { return null; } partitionManager = (PartitionManager)var2.next(); } while(partitionManager.getPartition().partition != partition); return partitionManager; }

### 重载ack／fail函数（ack 机制） ###

public void ack(Object msgId) { KafkaMessageId id = (KafkaMessageId)msgId; PartitionManager m = this._coordinator.getManager(id.partition); if (m != null) { m.ack(id.offset); } else { PartitionManager newManager = this.getManagerForPartition(id.partition.partition); if (newManager != null) { newManager.ack(id.offset); } } }     
     public void fail(Object msgId) { 
            KafkaMessageId id = (KafkaMessageId)msgId; PartitionManager m = this._coordinator.getManager(id.partition); if (m != null) { m.fail(id.offset); } else { PartitionManager newManager = this.getManagerForPartition(id.partition.partition); if (newManager != null) { newManager.fail(id.offset); } } }

##  ##

### 重载storm spout的取消激活函数 ###

public void deactivate() {         // spout被取消激活时，调用提交函数（见后文） this.commit(); }

###  ###

### 声明output fields ###

public void declareOutputFields(OutputFieldsDeclarer declarer) {         // 通过spoutConfig中的outputStreamId设置分流（默认default，else中的操作） if (!Strings.isNullOrEmpty(this._spoutConfig.outputStreamId)) { declarer.declareStream(this._spoutConfig.outputStreamId, this._spoutConfig.scheme.getOutputFields()); } else { declarer.declare(this._spoutConfig.scheme.getOutputFields()); } }

### 声明output fields ###

public Map<String, Object> getComponentConfiguration() {         // 设置当前组件的特殊配置信息（spout和bolt都是组件） Map<String, Object> configuration = super.getComponentConfiguration(); if (configuration == null) { configuration = new HashMap(); } String configKeyPrefix = "config."; ((Map)configuration).put(configKeyPrefix + "topics", this._spoutConfig.topic); StringBuilder zkServers = new StringBuilder(); String zkRoot;         // 获取并规范化zk连接信息（地址、IP） if (this._spoutConfig.zkServers != null && this._spoutConfig.zkServers.size() > 0) { Iterator var4 = this._spoutConfig.zkServers.iterator(); while(var4.hasNext()) { zkRoot = (String)var4.next(); zkServers.append(zkRoot + ":" + this._spoutConfig.zkPort + ","); } ((Map)configuration).put(configKeyPrefix + "zkServers", zkServers.toString()); }         // broker、partition信息，注意方式不同 BrokerHosts brokerHosts = this._spoutConfig.hosts; zkRoot = this._spoutConfig.zkRoot + "/" + this._spoutConfig.id; if (brokerHosts instanceof ZkHosts) { ZkHosts zkHosts = (ZkHosts)brokerHosts; ((Map)configuration).put(configKeyPrefix + "zkNodeBrokers", zkHosts.brokerZkPath); } else if (brokerHosts instanceof StaticHosts) { StaticHosts staticHosts = (StaticHosts)brokerHosts; GlobalPartitionInformation globalPartitionInformation = staticHosts.getPartitionInformation(); boolean useTopicNameForPath = globalPartitionInformation.getbUseTopicNameForPartitionPathId().booleanValue(); if (useTopicNameForPath) { zkRoot = zkRoot + "/" + this._spoutConfig.topic; } List<Partition> partitions = globalPartitionInformation.getOrderedPartitions(); StringBuilder staticPartitions = new StringBuilder(); StringBuilder leaderHosts = new StringBuilder(); Iterator var12 = partitions.iterator(); while(var12.hasNext()) { Partition partition = (Partition)var12.next(); staticPartitions.append(partition.partition + ","); leaderHosts.append(partition.host.host + ":" + partition.host.port).append(","); } ((Map)configuration).put(configKeyPrefix + "partitions", staticPartitions.toString()); ((Map)configuration).put(configKeyPrefix + "leaders", leaderHosts.toString()); } ((Map)configuration).put(configKeyPrefix + "zkRoot", zkRoot); return (Map)configuration; }

### 提交函数 ###

private void commit() { this._lastUpdateMs = System.currentTimeMillis(); Iterator var1 = this._coordinator.getMyManagedPartitions().iterator(); while(var1.hasNext()) {             // 使用manager的提交函数刷入offset信息 PartitionManager manager = (PartitionManager)var1.next(); manager.commit(); } }

### emit状态枚举类型 ###

static enum EmitState { EMITTED_MORE_LEFT, EMITTED_END, NO_EMITTED; private EmitState() { } } }