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这些比配，如果出现问题还可以重试