需求及选型
昨天在做简单日志采集分析，使用技术如下
- filebeat：做日志的收集
- kafka： 日志中转，缓冲及解耦作用
- storm： 做日志实时分析
后续会再加上其它，如
- elasticsearch ： 做搜索
- hbase : 存储
在此调研只使用最上面的三项。
kafka
其安装不做介绍。只需做简单配置，然后创建一个topic，命令如：
kafka-topics.sh --create --zookeeper 192.168.101.176:2181 --replication-factor 1 --partitions 1 --topic demolog1
filebeat
安装不做介绍。
修改filebeat.yml，注释掉其默认的elasticsearch的配置，修改要输出的log文件
在最后面增加输入到kafka的信息，如：
output.kafka:
enabled: true
hosts: ["192.168.101.176:9092"]
topic: demolog1
在此遇到一个问题：配置好ip，但发现不能找到kafka，其内部都是使用的域名，应该为：先通过ip查到域名，然后使用域名去查找，则这时找不到相应的主机了，这时只需要
在/etc/hosts里面配置好相应的域名即可。
storm
引用最新的kafka，及其client，最新的使用的是storm-kafka-client，网上已有的内容基本都是针对老版本kafka的，老版本使用的是storm-kafka，需要使用的pom如下：
<dependencies>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.1.1</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>1.1.1</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>0.11.0.1</version>
<exclusions>
<exclusion>
<artifactId>log4j</artifactId>
<groupId>log4j</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
这里需要把kafka中依赖的log4j去除，否则会报错，如果使用的是老版本的storm-kafka，则启动不了，是因为引入即有log4j，又有log4j2导致的。如果使用的storm-kafka-client，则会有以下
警告。
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/E:/repository/org/apache/logging/log4j/log4j-slf4j-impl/2.8.2/log4j-slf4j-impl-2.8.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/E:/repository/org/slf4j/slf4j-log4j12/1.7.25/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
此外还有一个问题是，虽然配置什么都正确，但是不能获取到kafka消息,最后发现是没有在host里面配置域名，虽然代码使用的是ip，但被解析后，再使用就不行了，增加host后解决。其原因应该和上面一样.
在此只从kafka中获得消息，不需要把消息写入到kafka，因此使用KafkaSpout
topology如下：
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
public class DemoTopology {
private static TopologyBuilder builder = new TopologyBuilder();
public static void main(String[] args) {
Config config = new Config();
config.setDebug(true);
config.setNumWorkers(1);
KafkaSpoutConfig kafkaSpoutConfig = KafkaSpoutConfig.builder("192.168.101.176:9092",
"demolog1")
.setProp("client.id","hello")
.setProp("group.id", "storm")
.setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
.build();
builder.setSpout("kafkaSpolt", new KafkaSpout<>(kafkaSpoutConfig), 1);
builder.setBolt("demoBolt", new DemoBolt(), 2).shuffleGrouping("kafkaSpolt");
if (args != null && args.length > 0) {
try {
config.setNumWorkers(1);
StormSubmitter.submitTopology(args[0], config,builder.createTopology());
} catch (Exception e) {
e.printStackTrace();
}
} else {
config.setMaxTaskParallelism(1);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("demo", config, builder.createTopology());
Utils.sleep(1000000000);
cluster.killTopology("demo");
cluster.shutdown();
}
}
}
则这时就可以在自己的Bolt中获取到消息了。
参考文档：http://storm.apache.org/releases/1.1.1/storm-kafka-client.html
