storm 读取不到对应的kafka数据

坑一：pom文件主要内容：注意里面 需要 使用 “exclusion”排除相关的依赖

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.7</maven.compiler.source>
    <maven.compiler.target>1.7</maven.compiler.target>
    <storm.version>1.1.1</storm.version>
    <kafka.version>0.9.0.0</kafka.version>
</properties>

<dependencies>

    <!--storm-core依赖-->
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>${storm.version}</version>
        <exclusions>
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>log4j-over-slf4j</artifactId>
            </exclusion>
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

    <!--storm-kafka 依赖-->
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-kafka</artifactId>
        <version>${storm.version}</version>
    </dependency>

    <!-- kafka 依赖-->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>${kafka.version}</version>
        <exclusions>
            <exclusion>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
            </exclusion>
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
            </exclusion>
        </exclusions>
    </dependency>


    <dependency>
        <groupId>commons-io</groupId>
        <artifactId>commons-io</artifactId>
        <version>2.4</version>
    </dependency>

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>${kafka.version}</version>
    </dependency>
</dependencies>

坑二： input.getBinaryByField(“bytes”); 里面一定要写成bytes，这是上游kafkaSpout 传递过来，源码中也可以看到。
对应位置如下图

业务代码体现：
public void execute(Tuple input) {
try {
byte[] bytes = input.getBinaryByField(“bytes”);
String value = new String(bytes);
System.out.println(“value ” + value);
this.collector.ack(input);
} catch (Exception e) {
e.printStackTrace();
this.collector.fail(input);
}
}

坑三：本地测试是，一直接收不到kafkaSpout发送过来的消息：
1）问题是已经连接上了kafka，也读到了对应的分区
2）推断可能是上游的数据发送不过来—》 可能原因shuffleGrouping时 的参数传递错误。
3）最终发现 原来就是SPOUT_ID 获取错了
应该将下面代码中的

     String SPOUT_ID = kafkaSpout.getClass().getSimpleName()
     替换成
     String SPOUT_ID = KafkaSpout.class.getSimpleName();
     即可。



         // kafka 使用的zk hosts
        BrokerHosts hosts = new ZkHosts("hadoop000:2181");
//        指定的kafak的一个根目录，存储的是kafkaSpout读取数据的位置信息（offset）
        SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName, "/" + topicName, UUID.randomUUID().toString());

        spoutConfig.startOffsetTime = kafka.api.OffsetRequest.LatestTime(); // 设置从最近的消息开始消费

        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
        String SPOUT_ID = kafkaSpout.getClass().getSimpleName();
        builder.setSpout(SPOUT_ID, kafkaSpout);

        String BOLD_ID = LogProcessBolt.class.getSimpleName();
        builder.setBolt(BOLD_ID, new LogProcessBolt()).shuffleGrouping(SPOUT_ID);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("StormToKafkaTopology", new Config(), builder.createTopology());

坑四： storm重复消费kafak数据：

官网解释如下：  

代码中配置为如下即可
   SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName, "/" + topicName, UUID.randomUUID().toString());

  spoutConfig.startOffsetTime = kafka.api.OffsetRequest.LatestTime(); // 设置从最近的消息开始消费

坑五： storm消费数据，ack，fail这些比配，如果出现问题还可以重试