首页 博客列表 精选博客 智能问答 chat-GPT 关于我
storm-kafka源码走读之KafkaSpout
发布日期：2021-07-12 08:49:14 浏览次数：3 分类：技术文章

本文共 8553 字，大约阅读时间需要 28 分钟。

from:

（五）storm-kafka源码走读之KafkaSpout

原创 2014年11月08日 14:09:06
  • 标签：
  •  /
  •  /
  •  /
  • 3458

现在开始介绍KafkaSpout源码了。

开始时，在open方法中做一些初始化，

 

[java] 
 
 
  1. ........................  
  2.   
  3.         _state = new ZkState(stateConf);  
  4.   
  5.         _connections = new DynamicPartitionConnections(_spoutConfig, KafkaUtils.makeBrokerReader(conf, _spoutConfig));  
  6.   
  7.         // using TransactionalState like this is a hack  
  8.         int totalTasks = context.getComponentTasks(context.getThisComponentId()).size();  
  9.         if (_spoutConfig.hosts instanceof StaticHosts) {  
  10.             _coordinator = new StaticCoordinator(_connections, conf, _spoutConfig, _state, context.getThisTaskIndex(), totalTasks, _uuid);  
  11.         } else {  
  12.             _coordinator = new ZkCoordinator(_connections, conf, _spoutConfig, _state, context.getThisTaskIndex(), totalTasks, _uuid);  
  13.         }  
  14. ............  
前后省略了一些代码，关于metric这系列暂时不介绍。主要是初始化Zookeeper连接zkstate，把kafka Partition 与broker关系对应起来（初始化DynamicPartitionConnections），在DynamicPartitionConnections构造函数需要传入一个brokerReader，我们是zkHosts，看KafkaUtils代码就知道采用的是ZkBrokerReader，来看下ZkBrokerReader的构造函数代码

 

 

[java] 
 
 
  1. public ZkBrokerReader(Map conf, String topic, ZkHosts hosts) {  
  2.         try {  
  3.             reader = new DynamicBrokersReader(conf, hosts.brokerZkStr, hosts.brokerZkPath, topic);  
  4.             cachedBrokers = reader.getBrokerInfo();  
  5.             lastRefreshTimeMs = System.currentTimeMillis();  
  6.             refreshMillis = hosts.refreshFreqSecs * 1000L;  
  7.         } catch (java.net.SocketTimeoutException e) {  
  8.             LOG.warn("Failed to update brokers", e);  
  9.         }  
  10.   
  11.     }  
有一个refreshMillis参数，这个参数是定时更新zk中partition的信息，

 

 

[java] 
 
 
  1. //ZkBrokerReader  
  2.     @Override  
  3.     public GlobalPartitionInformation getCurrentBrokers() {  
  4.         long currTime = System.currentTimeMillis();  
  5.         if (currTime > lastRefreshTimeMs + refreshMillis) { // 当前时间大于和上次更新时间之差大于refreshMillis  
  6.             try {  
  7.                 LOG.info("brokers need refreshing because " + refreshMillis + "ms have expired");  
  8.                 cachedBrokers = reader.getBrokerInfo();  
  9.                 lastRefreshTimeMs = currTime;  
  10.             } catch (java.net.SocketTimeoutException e) {  
  11.                 LOG.warn("Failed to update brokers", e);  
  12.             }  
  13.         }  
  14.         return cachedBrokers;  
  15.     }  
  16.     // 下面是调用DynamicBrokersReader 的代码  
  17.     /** 
  18.      * Get all partitions with their current leaders 
  19.      */  
  20.     public GlobalPartitionInformation getBrokerInfo() throws SocketTimeoutException {  
  21.       GlobalPartitionInformation globalPartitionInformation = new GlobalPartitionInformation();  
  22.         try {  
  23.             int numPartitionsForTopic = getNumPartitions();  
  24.             String brokerInfoPath = brokerPath();  
  25.             for (int partition = 0; partition < numPartitionsForTopic; partition++) {  
  26.                 int leader = getLeaderFor(partition);  
  27.                 String path = brokerInfoPath + "/" + leader;  
  28.                 try {  
  29.                     byte[] brokerData = _curator.getData().forPath(path);  
  30.                     Broker hp = getBrokerHost(brokerData);  
  31.                     globalPartitionInformation.addPartition(partition, hp);  
  32.                 } catch (org.apache.zookeeper.KeeperException.NoNodeException e) {  
  33.                     LOG.error("Node {} does not exist ", path);  
  34.                 }  
  35.             }  
  36.         } catch (SocketTimeoutException e) {  
  37.                     throw e;  
  38.         } catch (Exception e) {  
  39.             throw new RuntimeException(e);  
  40.         }  
  41.         LOG.info("Read partition info from zookeeper: " + globalPartitionInformation);  
  42.         return globalPartitionInformation;  
  43.     }  
GlobalPartitionInformation是一个Iterator类，存放了paritition与broker之间的对应关系，DynamicPartitionConnections中维护Kafka Consumer与parittion之间的关系，每个Consumer读取哪些paritition信息。这个COnnectionInfo信息会在storm.kafka.ZkCoordinator中会被初始化和更新，需要提到的一点是一个KafkaSpout包含一个SimpleConsumer

 

 

[java] 
 
 
  1. //storm.kafka.DynamicPartitionConnections  
  2.     static class ConnectionInfo {  
  3.         SimpleConsumer consumer;  
  4.         Set<Integer> partitions = new HashSet();  
  5.   
  6.         public ConnectionInfo(SimpleConsumer consumer) {  
  7.             this.consumer = consumer;  
  8.         }  
  9.     }  

 

 

再看ZkCoordinator类，看其构造函数

 

[java] 
 
 
  1. //storm.kafka.ZkCoordinator  
  2.     public ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId, DynamicBrokersReader reader) {  
  3.         _spoutConfig = spoutConfig;  
  4.         _connections = connections;  
  5.         _taskIndex = taskIndex;  
  6.         _totalTasks = totalTasks;  
  7.         _topologyInstanceId = topologyInstanceId;  
  8.         _stormConf = stormConf;  
  9.         _state = state;  
  10.         ZkHosts brokerConf = (ZkHosts) spoutConfig.hosts;  
  11.         _refreshFreqMs = brokerConf.refreshFreqSecs * 1000;  
  12.         _reader = reader;  
  13.     }  
_refreshFreqMs就是定时更新zk partition到本地的操作，在kafkaSpout中nextTuple方法中每次都会去调用ZkCoordinator的getMyManagedPartitions方法。该方法根据_refreshFreqMs参数定时更新partition信息

 

 

[java] 
 
 
  1. //storm.kafka.ZkCoordinator  
  2.     @Override  
  3.     public List<PartitionManager> getMyManagedPartitions() {  
  4.         if (_lastRefreshTime == null || (System.currentTimeMillis() - _lastRefreshTime) > _refreshFreqMs) {  
  5.             refresh();  
  6.             _lastRefreshTime = System.currentTimeMillis();  
  7.         }  
  8.         return _cachedList;  
  9.     }  
  10.   
  11.     @Override  
  12.     public void refresh() {  
  13.         try {  
  14.             LOG.info(taskId(_taskIndex, _totalTasks) + "Refreshing partition manager connections");  
  15.             GlobalPartitionInformation brokerInfo = _reader.getBrokerInfo();  
  16.             List<Partition> mine = KafkaUtils.calculatePartitionsForTask(brokerInfo, _totalTasks, _taskIndex);  
  17.   
  18.             Set<Partition> curr = _managers.keySet();  
  19.             Set<Partition> newPartitions = new HashSet<Partition>(mine);  
  20.             newPartitions.removeAll(curr);  
  21.   
  22.             Set<Partition> deletedPartitions = new HashSet<Partition>(curr);  
  23.             deletedPartitions.removeAll(mine);  
  24.   
  25.             LOG.info(taskId(_taskIndex, _totalTasks) + "Deleted partition managers: " + deletedPartitions.toString());  
  26.   
  27.             for (Partition id : deletedPartitions) {  
  28.                 PartitionManager man = _managers.remove(id);  
  29.                 man.close();  
  30.             }  
  31.             LOG.info(taskId(_taskIndex, _totalTasks) + "New partition managers: " + newPartitions.toString());  
  32.   
  33.             for (Partition id : newPartitions) {  
  34.                 PartitionManager man = new PartitionManager(_connections, _topologyInstanceId, _state, _stormConf, _spoutConfig, id);  
  35.                 _managers.put(id, man);  
  36.             }  
  37.   
  38.         } catch (Exception e) {  
  39.             throw new RuntimeException(e);  
  40.         }  
  41.         _cachedList = new ArrayList<PartitionManager>(_managers.values());  
  42.         LOG.info(taskId(_taskIndex, _totalTasks) + "Finished refreshing");  
  43.     }  
其中每个Consumer分配partition的算法是KafkaUtils.calculatePartitionsForTask(brokerInfo, _totalTasks, _taskIndex);

 

主要做的工作就是获取并行的task数，与当前partition做比较，得出一个COnsumer要负责哪些parititons的读取，具体算法去kafka文档吧

以上在KafkaSpout中做完了初始化操作，下面开始取数据发射数据了，来看nextTuple方法

 

[java] 
 
 
  1. // storm.kafka.KafkaSpout  
  2.     @Override  
  3.     public void nextTuple() {  
  4.         List<PartitionManager> managers = _coordinator.getMyManagedPartitions();  
  5.         for (int i = 0; i < managers.size(); i++) {  
  6.   
  7.             try {  
  8.                 // in case the number of managers decreased  
  9.                 _currPartitionIndex = _currPartitionIndex % managers.size();  
  10.                 EmitState state = managers.get(_currPartitionIndex).next(_collector);  
  11.                 if (state != EmitState.EMITTED_MORE_LEFT) {  
  12.                     _currPartitionIndex = (_currPartitionIndex + 1) % managers.size();  
  13.                 }  
  14.                 if (state != EmitState.NO_EMITTED) {  
  15.                     break;  
  16.                 }  
  17.             } catch (FailedFetchException e) {  
  18.                 LOG.warn("Fetch failed", e);  
  19.                 _coordinator.refresh();  
  20.             }  
  21.         }  
  22.   
  23.         long now = System.currentTimeMillis();  
  24.         if ((now - _lastUpdateMs) > _spoutConfig.stateUpdateIntervalMs) {  
  25.             commit();  
  26.         }  
  27.     }  
看完上述代码可知，所有的操作都是在PartitionManager中进行的，PartitionManager中会读取message信息，然后进行发射，主要逻辑在PartitionManager的next方法中

 

 

[java] 
 
 
  1. //returns false if it's reached the end of current batch  
  2.     public EmitState next(SpoutOutputCollector collector) {  
  3.         if (_waitingToEmit.isEmpty()) {  
  4.             fill();  
  5.         }  
  6.         while (true) {  
  7.             MessageAndRealOffset toEmit = _waitingToEmit.pollFirst();  
  8.             if (toEmit == null) {  
  9.                 return EmitState.NO_EMITTED;  
  10.             }  
  11.             Iterable<List<Object>> tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg);  
  12.             if (tups != null) {  
  13.                 for (List<Object> tup : tups) {  
  14.                     collector.emit(tup, new KafkaMessageId(_partition, toEmit.offset));  
  15.                 }  
  16.                 break;  
  17.             } else {  
  18.                 ack(toEmit.offset);  
  19.             }  
  20.         }  
  21.         if (!_waitingToEmit.isEmpty()) {  
  22.             return EmitState.EMITTED_MORE_LEFT;  
  23.         } else {  
  24.             return EmitState.EMITTED_END;  
  25.         }  
  26.     }  
如果_waitingToEmit列表为空，则去读取msg，然后进行逐条发射，每发射一条，break一下，返回EMIT_MORE_LEFT给KafkaSpout的nextTuple方法中，，然后进行判断是否该paritition读取的一次读取的message buffer size是否已发射完毕，如果发射完毕就进行下一个partition 数据读取和发射，

 

注意的一点是，并不是一次把该partition的所有待发射的msg都发射完再commit offset到zk，而是发射一条，判断一下是否到了该commit的时候了（开始时设置的定时commit时间间隔），笔者认为这样做的原因是为了好控制fail

KafkaSpout中的ack，fail，commit操作全部交给了PartitionManager来做，看代码

 

[java] 
 
 
  1. @Override  
  2.     public void ack(Object msgId) {  
  3.         KafkaMessageId id = (KafkaMessageId) msgId;  
  4.         PartitionManager m = _coordinator.getManager(id.partition);  
  5.         if (m != null) {  
  6.             m.ack(id.offset);  
  7.         }  
  8.     }  
  9.   
  10.     @Override  
  11.     public void fail(Object msgId) {  
  12.         KafkaMessageId id = (KafkaMessageId) msgId;  
  13.         PartitionManager m = _coordinator.getManager(id.partition);  
  14.         if (m != null) {  
  15.             m.fail(id.offset);  
  16.         }  
  17.     }  
  18.   
  19.     @Override  
  20.     public void deactivate() {  
  21.         commit();  
  22.     }  
  23.   
  24.     @Override  
  25.     public void declareOutputFields(OutputFieldsDeclarer declarer) {  
  26.         declarer.declare(_spoutConfig.scheme.getOutputFields());  
  27.     }  
  28.   
  29.     private void commit() {  
  30.         _lastUpdateMs = System.currentTimeMillis();  
  31.         for (PartitionManager manager : _coordinator.getMyManagedPartitions()) {  
  32.             manager.commit();  
  33.         }  
  34.     }  
所以PartitionManager是KafkaSpout的核心，很晚了，都3点多了，后续会不上PartitionManager的分析，晚安

 

 

 

版权声明：本文为博主原创文章，未经博主允许不得转载。
    • 本文已收录于以下专栏：

转载于:https://www.cnblogs.com/wq3435/p/8001106.html

转载地址：https://blog.csdn.net/ankan7400/article/details/102088928 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！

上一篇：kafka系列之(3)——Coordinator与offset管理和Consumer Rebalance
下一篇：apache kafka系列之在zookeeper中存储结构

发表评论

最新留言

不错！
[***.144.177.141]2024年04月03日 07时43分12秒

关于作者

    喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！
-- 愿君每日到此一游！

推荐文章

output_buffering详细介绍 2019-04-27
php缓冲 output_buffering和ob_start 2019-04-27
php error_reporting 详解 2019-04-27
剖析PHP中的输出缓冲 2019-04-27
HTTP响应头不缓存 2019-04-27
phpize 2019-04-27
PHP安装eAccelerator 2019-04-27
PHP新的垃圾回收机制:Zend GC详解 2019-04-27
linux上使用strace查看C语言级别的php源码【一种方法】 2019-04-27
ACCEPT()和ACCEPT4() 2019-04-27
php内核探索方法与资源 2019-04-27
PHP安装扩展mcrypt以及相关依赖项 【PHP安装PECL扩展的方法】 2019-04-27
Javascript到PHP加密通讯的简单实现 2019-04-27
德国SNS交友/视频网站Poppen.de的技术架构分享 2019-04-27
UNIX环境编程 2019-04-27
一笔画问题【数据结构-图论】 2019-04-27
红黑树 2019-04-27
安装多个gcc 2019-04-27
Linux0.01内核根目录Makefile注释 2019-04-27
【CSDN2012年度博客之星】需要您的一票，感谢大家的支持 2019-04-27
白红宇的个人博客 - 记录点点滴滴的事 - 您是第 308051538 位访客
访问时间: 2024-04-26 13:00:00 访问IP: 3.143.9.115     Copyright © 2020 - 2023 blog.css8.cn 京ICP备2021015314号-1 手机版