Storm Spout

本文主要介绍了Storm Spout，并以KafkaSpout为例，进行了说明。

概念

数据源（Spout）是拓扑中数据流的来源。一般 Spout 会从一个外部的数据源读取元组然后将他们发送到拓扑中。根据需求的不同，Spout 既可以定义为可靠的数据源，也可以定义为不可靠的数据源。一个可靠的 Spout 能够在它发送的元组处理失败时重新发送该元组，以确保所有的元组都能得到正确的处理；相对应的，不可靠的 Spout 就不会在元组发送之后对元组进行任何其他的处理。

一个 Spout 可以发送多个数据流。为了实现这个功能，可以先通过 OutputFieldsDeclarer 的 declareStream 方法来声明定义不同的数据流，然后在发送数据时在 SpoutOutputCollector 的 emit 方法中将数据流 id 作为参数来实现数据发送的功能。

Spout 中的关键方法是 nextTuple。顾名思义，nextTuple 要么会向拓扑中发送一个新的元组，要么会在没有可发送的元组时直接返回。需要特别注意的是，由于 Storm 是在同一个线程中调用所有的 Spout 方法，nextTuple 不能被 Spout 的任何其他功能方法所阻塞，否则会直接导致数据流的中断（关于这一点，阿里的 JStorm 修改了 Spout 的模型，使用不同的线程来处理消息的发送，这种做法有利有弊，好处在于可以更加灵活地实现 Spout，坏处在于系统的调度模型更加复杂，如何取舍还是要看具体的需求场景吧——译者注）。

Spout 中另外两个关键方法是 ack 和 fail，他们分别用于在 Storm 检测到一个发送过的元组已经被成功处理或处理失败后的进一步处理。注意，ack 和 fail 方法仅仅对上述“可靠的” Spout 有效。

实现

在实现Spout的时候，有两种常用的方式：

implements IRichSpout
extends BaseRichSpout

IRichSpout

从上图看出，IRchSpout 继承了ISpout 和 IComponent这两个接口，所以一共有9个函数需要实现。

open: 环境初始化，调用open函数。
close: 当ISpout停止的时候，进行调用，但是并不一定保证成功，因为集群是调用kill -9 来停止程序的.
active: 激活spout，将spout的状态从deactive转变为active，紧接着spout就会调用nextTuple。
deactive：关闭spout。
nextTuple： Spout向下游发射一组tuple。
ack: storm确认spout发射出的id为msgId的tuple已经被处理完毕。
fail: storm 确认spout发射出的id为msgId的tuple在下游处理失败了。
declareOutputFields(OutputFieldsDeclarer declarer)：声明当前Spout要发送的stram的field的名字。
getComponentConfiguration：在component中声明配置信息。只有一些以"topology."开头的配置才会被重写。并且这些配置信息也可以通过TopologyBuilder来覆盖。

BaseRichSpout

BaseRichSpout是一个抽象类，它实现了IRichSpout的部分接口。如果业务不要求实现这些接口的时候，可以使用BaseRichSpout。

public abstract class BaseRichSpout extends BaseComponent implements IRichSpout {
    @Override
    public void close() {
    }

    @Override
    public void activate() {
    }

    @Override
    public void deactivate() {
    }

    @Override
    public void ack(Object msgId) {
    }

    @Override
    public void fail(Object msgId) {
    }
}

举例KafkaSpout

KafkaSpoout的作用是从kafka中读取数据，然后发送给下游进行处理。
回忆一下，一般消费kafka的流程是：

创建一个consumer实例。
订阅tpoic。
消费数据。
进行处理。
commit offset。

具体的使用方式看这里。

在Storm中，它的使用方式如下：

      String bootstrapServers = projectProperties.getProperty("kafkaBootstrapServers");
        String[] topic = projectProperties.getProperty("kafkaConsumerTopic").split(",");
        String consumerGroupId = projectProperties.getProperty("kafkaConsumerGroupId");
        KafkaSpoutConfig<String, String> kafkaSpoutConfig = KafkaSpoutConfig.builder(bootstrapServers, topic)
                .setGroupId(consumerGroupId)
                .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_LATEST)
                .build();
    ....
    final TopologyBuilder tp = new TopologyBuilder();
    tp.setSpout("kafkaSpout", new KafkaSpout<String, String>(kafkaSpoutConfig));

首先创建一个KafkaSpoutConfig，这个里面包含了相关的配置。
然后创建KafkaSpout实例。

简要介绍下KafkaSpoutConfig，它不仅包含了kafka consumer的配置，还包含了 Kafka spout 的一些配置。

    // Kafka consumer configuration
    private final Map<String, Object> kafkaProps;
    private final Subscription subscription;
    private final SerializableDeserializer<K> keyDes;
    private final Class<? extends Deserializer<K>> keyDesClazz;
    private final SerializableDeserializer<V> valueDes;
    private final Class<? extends Deserializer<V>> valueDesClazz;
    private final long pollTimeoutMs;

    // Kafka spout configuration
    private final RecordTranslator<K, V> translator;
    private final long offsetCommitPeriodMs;
    private final int maxUncommittedOffsets;
    private final FirstPollOffsetStrategy firstPollOffsetStrategy;
    private final KafkaSpoutRetryService retryService;
    private final long partitionRefreshPeriodMs;
    private final boolean emitNullTuples;

KafkaSpoout 除了实现了从kafka中获取数据，然后emit之外，还实现了retry机制。

如果要从kafka中获取数据，就要初始化一个KafkaConsumer对象。这个过程是在构造函数中实现的：

    public KafkaSpout(KafkaSpoutConfig<K, V> kafkaSpoutConfig) {
        this(kafkaSpoutConfig, new KafkaConsumerFactoryDefault<K, V>());
    }

KafkaConsumerFactoryDefault 这个类的作用就是创建一个KafkaConsumer 对象。

在Spout中，第一个调用的是open函数。这个函数对所有需要的变量进行初始化。这里的变量太多了，如果列举会太专注细节，而忽视了流程。
在调用 open 函数之后，需要调用 active 函数启动spout，并订阅topic。

    @Override
    public void activate() {
        try {
            subscribeKafkaConsumer();
        } catch (InterruptException e) {
            throwKafkaConsumerInterruptedException();
        }
    }

active 之后，就是调用nextTuple函数，从kafka中poll数据，然后进行发送。每次发送完一个tuple后，就会在emitted 这个set中添加对应的记录,方便后续的追踪。


    @Override
    public void nextTuple() {
        try {
         //如果设置了自动提交，或者距离上次提交时间已经过了指定时间
                if (commit()) {
                    commitOffsetsForAckedTuples();
                }
                
                if (poll()) {
                    try {
                        setWaitingToEmit(pollKafkaBroker());
                    } catch (RetriableException e) {
                        LOG.error("Failed to poll from kafka.", e);
                    }
                }

                if (waitingToEmit()) {
                    emit();
                }
                
                ....
        } catch (InterruptException e) {
            throwKafkaConsumerInterruptedException();
        }
    }

后续的bolt按照预期处理完对应的tuple后，会进行ack。这时会调用ack函数。ack函数会将对应msg从emitted中去除。

    public void ack(Object messageId) {
        final KafkaSpoutMessageId msgId = (KafkaSpoutMessageId) messageId;
        if (!emitted.contains(msgId)) {
            .....
        } else {
            .....
            emitted.remove(msgId);
        }
    }

如果后续的bolt处理消息失败了，就会调用fail函数.而fail函数会进行重试。

    public void fail(Object messageId) {
        final KafkaSpoutMessageId msgId = (KafkaSpoutMessageId) messageId;
        if (!emitted.contains(msgId)) {
            .....
        }
        emitted.remove(msgId);
        msgId.incrementNumFails();
        if (!retryService.schedule(msgId)) {
            .....
            ack(msgId);
        }
    }

等到消息处理完了之后，就需要调用 deactive 函数，deactive函数就是commit offset 并 close comsumer。 close 函数的实现和 deactive函数的实现一模一样。

    @Override
    public void close() {
        try {
            shutdown();
        } catch (InterruptException e) {
            throwKafkaConsumerInterruptedException();
        }
    }

    private void shutdown() {
        try {
            if (!consumerAutoCommitMode) {
                commitOffsetsForAckedTuples();
            }
        } finally {
            //remove resources
            kafkaConsumer.close();
        }
    }

一些细节

上面描述了KafkaSpout的大体流程，这里记录下其它的实现细节。

KafkaConsumerFactory

在创建kafka的comsumer对象的时候，使用了接口KafkaConsumerFactory，而这个接口只有一个实现KafkaConsumerFactoryDefault。

FirstPollOffsetStrategy

在第一次重kafka中读取数据的时候，spout提供了4种不同的策略。

/**
     * <li>EARLIEST means that the kafka spout polls records starting in the first offset of the partition, regardless of previous commits</li>
     * <li>LATEST means that the kafka spout polls records with offsets greater than the last offset in the partition, regardless of previous commits</li>
     * <li>UNCOMMITTED_EARLIEST means that the kafka spout polls records from the last committed offset, if any.
     * If no offset has been committed, it behaves as EARLIEST.</li>
     * <li>UNCOMMITTED_LATEST means that the kafka spout polls records from the last committed offset, if any.
     * If no offset has been committed, it behaves as LATEST.</li>
*/
    public static enum FirstPollOffsetStrategy {
        EARLIEST,
        LATEST,
        UNCOMMITTED_EARLIEST,
        UNCOMMITTED_LATEST }

Subscription

Subscription 封装了 kafka consumer 的 subscribe，然后提供了两个实现：

NamedSubscription
PatternSubscription

两者的区别就是subscribe 的方式不同。

public class PatternSubscription extends Subscription {
    .....
    @Override
    public <K, V> void subscribe(KafkaConsumer<K, V> consumer, ConsumerRebalanceListener listener, TopologyContext unused) {
        consumer.subscribe(pattern, listener);
        LOG.info("Kafka consumer subscribed topics matching wildcard pattern [{}]", pattern);
        
        // Initial poll to get the consumer registration process going.
        // KafkaSpoutConsumerRebalanceListener will be called following this poll, upon partition registration
        consumer.poll(0);
    }
    .....

}

为什么要做这层封装，我猜的是要通过 consumer.poll(0)来触发KafkaSpoutConsumerRebalanceListener。

KafkaSpoutConsumerRebalanceListener

Kafka 在增减consumer, partition, broker的时候会触发rebalance。 rebalance 之后， consumer对应的partition就会发生变化。这个时候要确保两件事情，第一是rebalance之前要commit 当前partition 消费的offset，第二是从新的 partition 获取当前的offset。而这些，都是通过实现ConsumerRebalanceListener达到目的。
在Storm中，就是：

    private class KafkaSpoutConsumerRebalanceListener implements ConsumerRebalanceListener {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

            if (!consumerAutoCommitMode && initialized) {
                initialized = false;
                //提交所有的已经ack的tuple的offset
                commitOffsetsForAckedTuples();
            }
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            //根据partitions进行初始化，包括获取当前的offset，更新其他变量..
            initialize(partitions);
        }

KafkaSpoutMessageId

    private transient TopicPartition topicPart;
    private transient long offset;
    private transient int numFails = 0;
    private boolean emitted;

FailedMsgRetryManager

Storm对失败的消息如何处理，有下面的几个问题？

消息处理失败了，应该怎么弄？
消息重复处理失败了，应该怎么弄？
消息重复处理成功了，应该怎么弄？
怎么获取将要处理的消息？
怎么知道这条消息是否需要发送？

KafkaSpoutRetryService 接口和它的实现KafkaSpoutRetryExponentialBackoff就是用来解决这个问题的。
在KafkaSpoutRetryService 设计了几种方法，包括添加、去掉处理失败的消息，查看当前的消息是否已经被添加和当前的消息是否达到了重新发送的条件。

KafkaSpoutRetryExponentialBackoff 实现了一种消息失败的处理方式。如果某条消息处理失败了，就会重试一定的次数，并且每次重试的时间按照指数时间增加。当然如果超过了最大的重试次数，KafkaSpou默认会将它ACK掉。

那么怎么实现的呢？

使用RetrySchedule 表示每条处理失败了消息，里面包含了：

private final KafkaSpoutMessageId msgId;
private long nextRetryTimeNanos;

这里面nextRetryTimeNanos表示下次重试的时间，如果这个时间小于当前的时间，就说明这条消息可以重试了。

KafkaSpoutRetryExponentialBackoff 使用了两个数据结构来保存每条要重试的数据，其中一个retrySchedules是一个treeSet，里面按照nextRetryTimeNanos从小到大进行排序。toRetryMsgs是一个HashSet，查找某条重试数据的状态。
```
private final Set<RetrySchedule> retrySchedules = new TreeSet<>(RETRY_ENTRY_TIME_STAMP_COMPARATOR);
private final Set<KafkaSpoutMessageId> toRetryMsgs = new HashSet<>();      // Convenience data structure to speedup lookups
```
消息处理失败了，应该怎么弄？

会增加这条消息的失败次数，然后调用schedule函数，这个函数会将消息添加到上面两个数据结构中。如果添加失败，就说明这个消息已经超过最大处理次数。否则的话，就会更新数据结构。
消息重复处理失败了，应该怎么弄？

处理过程同上，只不过会将retrySchedules和toRetryMsgs中对应的数据先删掉，然后再添加更新后的数据。
怎么获取将要处理的消息？

先通过retriableTopicPartitions来获取需要重试的消息的TopicPartition集合，然后重新从这些TopicPartition 中获取数据。
怎么知道这条消息是否需要发送？

遍历retrySchedules 进行查找。
consumer Poll 的时候，是拉取多条消息的，怎么保证某条消息处理失败了，重新拉取后，只向bolts中发送这条处理失败的消息？
1. nextTuple函数中，consumer poll一堆消息后，它会逐条发送，并且将已经发送的消息保存到 emitted 这个 Set 中。
2. 如果这条消息处理成功了，会将这条消息从emitted 中去掉，并保存到acked中。
3. 如果这条消息处理失败了，会将这条消息从emitted中去掉，添加到retryService中。
4. 在nextTuple重新拉取数据的时候，它会优先从需要retry的offset处开始拉取消息，这样就会重复拉取一些消息，所以，在emit的时候，会先从emitted 和acked 中查看是否包含了这条消息，如果不包含，就会发送。这样子发送的消息就只会有那条处理失败的消息了。
5. 当然这条消息如果多次失败，也会被标记为处理成功了。