（以下测试内容都是基于jdk1.8.0_66，操作系统为win10，仅供学习和测试，部分内容引用自官网资源，部分引用自《Learning Storm》）
源码已上传github：https://github.com/kyo-qin/tiger/tree/master/tiger-storm-common

1 启动zookeeper

  配置zk的端口为2181，通过命令行启动：

bin\zkServer.cmd

2 启动kafka

  配置kafka中的zk端口为2181，启动并创建主题，主题名为test：

bin\windows\kafka-server-start.bat config\server.properties
bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

3 编写storm模块

  笔者使用的storm版本为apache-storm-1.2.2

3.1 通过maven加载依赖

        <!-- storm核心依赖 -->
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>1.2.2</version>
        </dependency>
        <!-- storm-kafka连接客户端 -->
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-kafka-client</artifactId>
            <version>1.2.2</version>
        </dependency>
        <!-- kafka连接客户端 -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.0.0</version>
            <!-- <scope>${provided.scope}</scope> -->
        </dependency>

注：老版本使用的storm-kafka依赖已经被废弃，建议在以后使用storm-kafka-client依赖进行开发，老版本的storm-kafka依赖为：

<!--    <dependency> -->
<!--        <groupId>org.apache.storm</groupId> -->
<!--        <artifactId>storm-kafka</artifactId> -->
<!--        <version>1.2.2</version> -->
<!--    </dependency> -->

3.2 编写bolt处理类

  我们借用《learning storm》书上的例子，写一组将输入的单词拼成一句话的bolt类，每行输入一个单词，当输入符号“.”时，视为一句话结束。例如：

I
am
a
student
.

应解析为：

I am a student.

  核心代码为：

    private List<String> words = new ArrayList<String>();

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        // Get the word from the tuple
        String word = input.getString(0);
        if (StringUtils.isBlank(word)) {
            // ignore blank lines
            return;
        }
        logger.info("Received Word:" + word);
        System.out.println("Received Word:" + word);
        // add word to current list of words
        words.add(word);
        if (word.endsWith(".")) {
            // word ends with '.' which means this is the end
            // the SentenceBolt publishes a sentence tuple
            collector.emit(new Values(StringUtils.join(words, ' ')));
            // and reset the words list.
            words.clear();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("sentence"));
    }

  再编写一个bolt用于打印输出：

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        // get the sentence from the tuple and print it
        String sentence = input.getString(0);
        logger.info("Received Sentence: " + sentence);
        System.out.println("Received Sentence: " + sentence);
    }

3.2 配置spout

  新版本的KafkaSpout通过KafkaSpoutConfig类进行配置，KafkaSpoutConfig定义了kafka相关的环境、主题、重试策略、消费的初始偏移量等等参数。示例定义如下：

    protected KafkaSpoutConfig<String, String> newKafkaSpoutConfig() {
        return KafkaSpoutConfig.builder("localhost:9092", "test").setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup")
                .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 200).setRecordTranslator(JUST_VALUE_FUNC, new Fields("str"))
                .setRetry(newRetryService()).setOffsetCommitPeriodMs(10000).setFirstPollOffsetStrategy(LATEST)
                .setMaxUncommittedOffsets(250).build();
    }

其中JUST_VALUE_FUNC为kafka消息翻译函数，这边简单的将其输出：

    private static Func<ConsumerRecord<String, String>, List<Object>> JUST_VALUE_FUNC = new Func<ConsumerRecord<String, String>, List<Object>>() {
        @Override
        public List<Object> apply(ConsumerRecord<String, String> record) {
            return new Values(record.value());
        }
    };

newRetryService定义了重试策略：

    protected KafkaSpoutRetryService newRetryService() {
        return new KafkaSpoutRetryExponentialBackoff(new TimeInterval(500L, TimeUnit.MICROSECONDS), TimeInterval.milliSeconds(2),
                Integer.MAX_VALUE, TimeInterval.seconds(10));
    }

3.3 配置topology

  将上述bolt和spout以及配置类组合，配置topology：

    public StormTopology buildTopology() {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("KafkaSpout", new KafkaSpout<String, String>(newKafkaSpoutConfig()), 1);
        builder.setBolt("SentenceBolt", new SentenceBolt(), 1).globalGrouping("KafkaSpout");
        builder.setBolt("PrinterBolt", new PrinterBolt(), 1).globalGrouping("SentenceBolt");
        return builder.createTopology();
    }

3.4 本地测试运行

    public static void main(String[] args) {
        KafkaTopologyBasic kb = new KafkaTopologyBasic();
        StormTopology topology = kb.buildTopology();
        LocalCluster cluster = new LocalCluster();
        Config conf = new Config();
        cluster.submitTopology("KafkaToplogy", conf, topology);
        try {
            // Wait for some time before exiting
            System.out.println("Waiting to consume from kafka");
            Thread.sleep(300000);
        } catch (Exception exception) {
            System.out.println("Thread interrupted exception : " + exception);
        }
        // kill the KafkaTopology
        cluster.killTopology("KafkaToplogy");
        // shut down the storm test cluster
        cluster.shutdown();
    }

通过kafka命令行工具生产消息：

bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test

输入：

>I       
>am      
>a       
>student 
>.       

输出：

2018-08-15 17:11:30 [Thread-20-SentenceBolt-executor[3 3]] INFO  [org.tiger.storm.common.kafka.SentenceBolt] - Received Word:I
Received Word:I
2018-08-15 17:11:30 [Thread-20-SentenceBolt-executor[3 3]] INFO  [org.tiger.storm.common.kafka.SentenceBolt] - Received Word:am
Received Word:am
2018-08-15 17:11:32 [Thread-20-SentenceBolt-executor[3 3]] INFO  [org.tiger.storm.common.kafka.SentenceBolt] - Received Word:a
Received Word:a
2018-08-15 17:11:35 [Thread-20-SentenceBolt-executor[3 3]] INFO  [org.tiger.storm.common.kafka.SentenceBolt] - Received Word:student
Received Word:student
2018-08-15 17:11:35 [Thread-20-SentenceBolt-executor[3 3]] INFO  [org.tiger.storm.common.kafka.SentenceBolt] - Received Word:.
Received Word:.
2018-08-15 17:11:35 [Thread-18-PrinterBolt-executor[2 2]] INFO  [org.tiger.storm.common.kafka.PrinterBolt] - Received Sentence: I am a student .
Received Sentence: I am a student .