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