storm之spout

一、什么是spout

spout：喷嘴、喷口。即数据从这里发出。

spout是storm的数据来源，而spout的数据来源又是从其他地方，比如数据库或者消息中间件中流入的。

以Kafka为例，spout先从kafka中拉取数据，然后封装为一个tuple，发给下游的bolt进行处理。对于Kafka来说，spout是消费者；对于bolt来说spout是生产者。

为什么要用spout去拉取消息，而不是直接由bolt接收推送的数据呢，这中拉模式有什么好处呢？

如果，将数据直接推送给bolt，当数据量突然增加的时候，可能导致某一个bolt瘫痪，继而影响整个topology运行；而当没有数据的时候，整个topolog又处于空闲状态，浪费资源。而由spout去拉取消息则不会出现这样的问题。

二、KafkaSpout

KafkaSpout实现了从Kafka拉取数据为storm提供数据源。并且重新实现了ack机制。一般的我们通过简单的配置就可以使用了。

	//kafkaSpout配置
    private KafkaSpoutConfig<String, String> kafkaSpoutConfig() {
        final Fields outputFields = new Fields("topic", "partition", "offset", "timestamp", "key", "msg_from_kafka");
        KafkaSpoutConfig<String, String> config;
        //consumer的配置
        Properties props = new Properties();
        //默认由kafkaSpout进行ack后才提交（false），如果自动提交，则kafkaspout的ack失效，可能丢失或重复数据
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        KafkaSpoutRetryService kafkaSpoutRetryService = new KafkaSpoutRetryExponentialBackoff(
                TimeInterval.microSeconds(500),
                TimeInterval.milliSeconds(2),
                1,
                TimeInterval.seconds(10));

        config = KafkaSpoutConfig
                .builder("ip:9092", "topic_test")
                //首次消费消息的offset
                .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST)
                //最后一个参数为输出字段
                .setRecordTranslator((r) -> new Values(r.topic(), r.partition(), r.offset(), r.timestamp(), r.key(), r.value()), outputFields)
                //offset自动提交时间间隔，如果设置了enable.auto.commit=true则无效
                .setOffsetCommitPeriodMs(1_000)//1秒
                //达到这个值后向提交offset
                .setMaxUncommittedOffsets(1_000_000)//10万
                //group
                .setGroupId("test-w")
                //kafka consumer配置
                .setProp(props)
                .setRetry(kafkaSpoutRetryService)
                .build();
        return config;
    }

	//拓扑结构
    private StormTopology stormTopology() {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("spout", new ProducerSpout(kafkaSpoutConfig()), 1);
        builder.setBolt("bolt1", new BoltTest(), 1).shuffleGrouping("spout");
        return builder.createTopology();
    }

kafkaspout的所有配置项：

public static final long DEFAULT_POLL_TIMEOUT_MS = 200L;
public static final long DEFAULT_OFFSET_COMMIT_PERIOD_MS = 30000L;
public static final int DEFAULT_MAX_RETRIES = 2147483647;
public static final int DEFAULT_MAX_UNCOMMITTED_OFFSETS = 10000000;
public static final long DEFAULT_PARTITION_REFRESH_PERIOD_MS = 2000L;
public static final KafkaSpoutRetryService DEFAULT_RETRY_SERVICE = new KafkaSpoutRetryExponentialBackoff(TimeInterval.seconds(0L), TimeInterval.milliSeconds(2L), 2147483647, TimeInterval.seconds(10L));
public static final KafkaSpoutRetryService UNIT_TEST_RETRY_SERVICE = new KafkaSpoutRetryExponentialBackoff(TimeInterval.seconds(0L), TimeInterval.milliSeconds(0L), 2147483647, TimeInterval.milliSeconds(0L));
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;
private final RecordTranslator<K, V> translator;
private final long offsetCommitPeriodMs;
private final int maxUncommittedOffsets;
private final KafkaSpoutConfig.FirstPollOffsetStrategy firstPollOffsetStrategy;
private final KafkaSpoutRetryService retryService;
private final long partitionRefreshPeriodMs;
private final boolean emitNullTuples;

具体含义在后面会总结。

---

参考资料：

《storm技术内幕与大数据实战》