基于0.93版本Storm

首先，如果自己写KafkaSpout，该怎么办？有哪些地方需要考虑呢

1. 得实现Storm指定的接口。这样Storm才能够使用它。那么需要实现什么接口？需要提供什么功能给Storm调用呢？

2. 需要给spout的每个task指定任务，也就是把Kafka里的消息分配给spout task去读取。这时候，就会有以下问题：

• 是否一个KafkaSpout需要支持多个topic?鉴于每个topology里可以有多个spout，这样做没有必要，而且会带来较大的复杂性。
• 如何把partitions分给tasks? 这时候存在如下问题：
• 是否每个partition最多只分给一个task?理论上，我们可以将一个partition的消息给多个task处理，只要这些task区分自己负责的那部分消息就行，比如一个task读偶offset的消息，一个读奇offset的消息。
• 如何分才好。考虑到负载的平衡，而且要使得不同task间的任务不会冲突？比如，不会出现两个task读相同的消息的情况。
• 每个partition从何处开始读取？如何记录对当前Kafka topic的消费进度，使得在topology下线以后，这部分消息不会丢失，以便以后可以接着上次的消费过度处理。

3. 如何读取？怎么使用Kafka API读取消息？每次读多大量的消息？需要预读和缓冲吗？

4. 无法从Kafka读取消息时如何处理？在spout里重试?认为spout出现异常，交由Storm重新调度？

5. 当KafkaSpout的进度严重落后于Kafka消息的数量时该如何处理？当spout读取的速度太小，使得Kafka里未被处理的消息越来越多时如何处理？

6. 需要读取的消息不存在该如何处理？比如从Kafka取消息时，想要获取的消息已经由于存储时间过久，被Kafka删除了，该如何处理？

7. 一个启用了log compaction的topic会有何不同？

先列一下KafkaSpout的实现里的关键类，以便接下来分析代码时更好理解

KafkaSpout的open方法

每个Spout task会有一个KafkaSpout的实例。当这个task初始化时，Storm会调用KafkaSpout的open方法，初始化这个spout task的运行环境，包括

• a. 分配partiton给这个task
• b. 为分到的每个partition生成一个PartitionManager。PartitionManager对于每个partition的消息实现了Spout接口的ack, nextTuple, fail等主要功能。

关键代码如下：KafkaSpout的open方法主要用来为当前的spout task提供一个Coordinator.

//创建一个DynamicPartitionConnections，用于获取partition对应的SimpleConsumer

        _connections = new DynamicPartitionConnections(_spoutConfig, KafkaUtils.makeBrokerReader(conf, _spoutConfig));

        // using TransactionalState like this is a hack

        //总共有多少task

        int totalTasks = context.getComponentTasks(context.getThisComponentId()).size();

        if (_spoutConfig.hosts instanceof StaticHosts) {

            _coordinator = new StaticCoordinator(_connections, conf, _spoutConfig, _state, context.getThisTaskIndex(), totalTasks, _uuid);

        } else {

            _coordinator = new ZkCoordinator(_connections, conf, _spoutConfig, _state, context.getThisTaskIndex(), totalTasks, _uuid);

        }

其中，在KafkaConfig中使用StaticHosts还是ZkHosts对DynamicParitionConnections和Coordinator的行为都有影响。

DynamicPartitonConnections  为Partition提供SimpleConsumer

因为Kafka的每个SimpleConsumer都可以用于与一个broker通信，不管是否这些请求是针对同一个topic或partition。当一个broker作为多个partition的leader时，只需要为这一个broker建立一个SimpleConsumser，就可以用于消费这多个partition。所以需要DynamicPartitionConnection来管理partition与SimpleConsumser之间的对应关系，更好地复用。

• 当使用StaticHosts时，KafkaUtils.makeBrokerReader(conf, _spoutConfig)会生成一个StaticBrokerReader. 这个BrokerReader只会提供StaticHosts实例化时使用的分区信息。使得DynamicPartitionConnection的register(Partition partition)方法被调用时，只会返回同样的SimpleConsumer。
• 当使用ZkHosts时，KafkaUtils.makeBrokerReader(conf, _spoutConfig)会生成一个ZkBrokerReader。这个BrokerReader带有自动刷新功能，当两次对它的的getCurrentBrokers的调用间隔较长，它就会重新获取这个topic的GlobalParitionInformation，即重新获取分区和分区的leader。使得DynamicPartitionConnection的register(Partition partition)方法被调用时，有可能会重新获取最新的分区信息。

Coordinator 为task分配partition，并且为每个partition建立PartitionManager

　　Coordinator如何为task分配Partition?

无论是StaticCoordinator还是ZkCoordinator都是使用KafkaUtilsCalculatorPartitionsForTask方法来给task分配partitions

    public static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex) {

        Preconditions.checkArgument(taskIndex < totalTasks, "task index must be less that total tasks");

        List<Partition> partitions = partitionInformation.getOrderedPartitions();

        int numPartitions = partitions.size();

        if (numPartitions < totalTasks) {

            LOG.warn("there are more tasks than partitions (tasks: " + totalTasks + "; partitions: " + numPartitions + "), some tasks will be idle");

        }

        List<Partition> taskPartitions = new ArrayList<Partition>();

        for (int i = taskIndex; i < numPartitions; i += totalTasks) {

            Partition taskPartition = partitions.get(i);

            taskPartitions.add(taskPartition);

        }

        logPartitionMapping(totalTasks, taskIndex, taskPartitions);

        return taskPartitions;

    }

　　若一个task的index为a, 那么分给它的partition在所有partition中的index(如果用StaticHosts，并且只提供了部分partition，那么可能partition的index并不是partition id)为:

　　partitionIndex = a + k*totalTasks， k是正整数，且partitionIndex < numPartitions

• 当使用StaticHosts时，KafkaSpout会使用StaticCoordinator，这种Cooridnator的refresh方法什么都不会做。
• 当使用ZkHosts时，KafkaSpout会使用ZkCoordinator。这种Coordinator的refresh方法被调用时，它会通过BrokerReader获取最新的分区信息，重新为当前的task计算分区，然后为新的分区提供PartitionManager，从当前task的分区表时移除旧的分区，关闭旧的分区。注意，当某个分区的leader变更后，它对应的Partition实例的broker字段会和以前的不同，因此会认为是新的Partition。当这种Coordinator的getMyManagedPartitions方法被调用时，如果过太久没刷新，它就会调用refresh()方法，重新获取这个task对应的PartitionManager集合。

  public List<PartitionManager> getMyManagedPartitions() {
  
          if (_lastRefreshTime == null || (System.currentTimeMillis() - _lastRefreshTime) > _refreshFreqMs) {
  
              refresh();
  
              _lastRefreshTime = System.currentTimeMillis();
  
          }
  
          return _cachedList;
  
      }

  • 那么何时getMyManagedPartition会被调用呢？是在KafkaSpout的nextTuple方法被调用时。也就是每次nextTuple被调用, ZkCoordinator都会检查是否需要更新PartitionManager集合。

  • 如果partition的leader发生成了变更，而Coordinator没有刷新呢？此时，按照旧的leader获取消息，就抛出异常。而KafkaSpout的nextTuple方法会捕获异常，然后主动调用coordinator的refresh()方法获取新的PartitionManager集合。

KafkaSpout对于IRichSpout接口的实现

　　nextTuple方法的实现

 public void nextTuple() {

        List<PartitionManager> managers = _coordinator.getMyManagedPartitions();

        for (int i = 0; i < managers.size(); i++) {

            try {

                // in case the number of managers decreased

                _currPartitionIndex = _currPartitionIndex % managers.size();

                EmitState state = managers.get(_currPartitionIndex).next(_collector);

                if (state != EmitState.EMITTED_MORE_LEFT) {

                    _currPartitionIndex = (_currPartitionIndex + 1) % managers.size();

                }

                if (state != EmitState.NO_EMITTED) {

                    break;

                }

            } catch (FailedFetchException e) {

                LOG.warn("Fetch failed", e);

                _coordinator.refresh();

            }

        }

        long now = System.currentTimeMillis();

        if ((now - _lastUpdateMs) > _spoutConfig.stateUpdateIntervalMs) {

            commit();

        }

    }

首先，它会从coordinator处获取当前所管理的所有partition.然后试着从这些partition的消息中emit tuple, 由于可以采用schema解析Kafka的消息，使得一个消息对应多个tuple，所以这里每次试用nextTuple，可能实际上会emit多个tuple。这就带来了一个问题，如果一个 Kafka message生成多个tuple，那么是否这些tuple都被ack了，才认为这个Kafka消息处理完了呢？实际上，现在的KafkaSpout的实现里，只要其中有一个tuple失败了，就认为message失败了。

可以看到，代码里的for循环最多会循环manager.size()次，也就是它管理多少个partition，就最多循环几次。但实际上，只要有一个消息产生了tuple，for循环就会终止。也就是nextTuple被调用后，只要有一条消息被成功解析为tuple，它就不再继续处理消息，在按配置时间间隔记录下进度后，方法就执行完毕。nextTuple方法调用PartitionManager来emit tuple，根据PartitionManager的next方法返回的状态nextTuple的控制流程。PartitionManager的next方法最多只emit一条消息产生的所有tuple，先说一下这个next方法返回的状态的意义：

• NO_EMITTED 表示此次调用没有emit任何tuple。其它状态都是已经从一条消息emit了tuple，有可能处理了多条消息，但可能最初的消息没能解析成tuple，但只有一条消息解析成tuple，next方法就不会再处理消息。
• EMITTED_MORE_LEFT 表示已经处理了一个消息emit了一个或一些tuple, 但是这个partition还有消息已经被读取却还没有处理。
• EMITTED_END 表示已经从一个消息emit了一个或一些tuple，并且这个partition所有已经获取的消息都已经被处理了。

根据这些状态，KafkaSpout做出以下处理：

• 如果不是NO_EMITTED，也就是EMITTED_MORE_LEFT或者EMITTED_END，表示已经emit了tuple，所以就退出for循环，不再emit新的tuple.
• 如果不是EMITTED_MORE_LEFT，说明这个PartitionManager已读的消息都已进行了处理，下次就从另一个PartitionManager处获取消息，所以更新_currentPartitionIndex

不管是emit了tuple而退出循环, 或者把当前管理的partition循环了一遍之后还却没有emit任何消息而退出循环。nextTuple的最后都会检查是否需要在Zookeeper里记录进度。

KafkaSpout的ack, commit, fail方法的具体逻辑都由PartitionManager来实现。下一篇会详细进行分析。