需求及选型

昨天在做简单日志采集分析，使用技术如下

1. filebeat：做日志的收集
2. kafka： 日志中转，缓冲及解耦作用
3. 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