windows下如何实现kafka+ELK的日志系统-创新互联

这篇文章主要介绍windows下如何实现kafka+ELK的日志系统,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!

目前成都创新互联已为成百上千的企业提供了网站建设、域名、网络空间、网站托管、企业网站设计、杭州网站维护等服务,公司将坚持客户导向、应用为本的策略,正道将秉承"和谐、参与、激情"的文化,与客户和合作伙伴齐心协力一起成长,共同发展。

用到的软件:zookeeper、kafka、logstash(6.3.2版本)、ES(6.3.2版本)、Kibana(6.3.2版本)。具体安装步骤不在此说明,基本都是下载解压,改一下配置文件,即可使用。(以下所述均在Windows下)
1、zookeeper:
kafka中自带zookeeper,可以不用装zookeeper,如果想自己另装,需配置环境变量,如下:
ZOOKEEPER_HOME  =>  D:\nomalAPP\zookeeper-3.4.13
path 里面加入  %ZOOKEEPER_HOME%\bin
如果配置好以后,在cmd里运行zkserver报找不到java错误的话,可能是java环境变量的位置放置有问题,可以将path里面配置的java环境位置移到最前面。

2、kafka:
kafka如果启动报找不到java的错误,原因在于kafka-run-class.bat的179行,找到里面的  %CLASSPATH% ,将其用双引号括起来。即改为:set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp "%CLASSPATH%" %KAFKA_OPTS% %*
kafka的配置文件server.properties里面要修改的:
日志所放置的目录(可以选择用默认的):log.dirs=D:/nomalAPP/kafka_2.12-2.0.0/kafka-logs,
连接zookeeper的ip和端口:zookeeper.connect=localhost:2181
其他的均可用默认配置。
启动命令:cmd锁定安装目录,然后  .\bin\windows\kafka-server-start.bat .\config\server.properties

3、logstash:
在config目录下,logstash.yml里面,如果想启动多个,在里面可以配置 http.port=9600-9700,如果没有配置这一项,会使用默认的9600端口。
在config目录下,增加一个logstash.conf的配置文件,用来配置数据源,和数据过滤,数据输出的位置,如下:

input {
            #   ==>    用kafka的日志作为数据源
   kafka{  
     ····#(kafka的IP和端口)
        bootstrap_servers => "10.8.22.15:9092"  
        # (在有多个相同类型的数据源时,需要配置)
        client_id => "test1"  
        # (消费者分组,可以通过组 ID 去指定,不同的组之间消费是相互不受影响的,相互隔离)
        group_id => "test1"   
        auto_offset_reset => "latest"    
        # (消费者线程个数)
        consumer_threads => 5   
        # (在输出消息的时候会输出自身的信息包括:消费消息的大小, topic 来源以及 consumer 的 group 信息)
        decorate_events => true    
        # (主题)
        topics => ["bas-binding-topic"]    
        # (用于ES索引)
        type => "bas-binding-topic"  
        # (数据格式)
        codec => "json"     
        tags => ["bas-binding-topic"]
      }  
      kafka{  
        bootstrap_servers => "10.8.22.15:9092"  
        client_id => "test2"  
        group_id => "test2"  
        auto_offset_reset => "latest"  
        consumer_threads => 5
        #decorate_events => true  
        topics => ["bas-cus-request-topic"]  
        type => "bas-cus-request-topic"  
        codec => "json"
        tags => ["bas-cus-request-topic"]
      }  
}

filter {
    filter插件负责过滤解析input读取的数据,可以用grok插件正则解析数据,
    date插件解析日期,json插件解析json等等
}

output {
    # 输出到ES
    if "bas-binding-topic" in [tags] {
       elasticsearch{  
             # ES的IP和端口
            hosts => ["localhost:9201"]  
            # 索引名称:主题+时间
            index => "bas-binding-topic-%{+YYYY.MM.dd+HH:mm:ss}"  
            timeout => 300
        }
        stdout {
            codec => rubydebug
         }
  }
  if "bas-cus-request-topic" in [tags] {
       elasticsearch{  
            hosts => ["localhost:9201"]  
            index => "bas-cus-request-topic-%{+YYYY.MM.dd+HH:mm:ss}"  
            timeout => 300
        }
        stdout {
            codec => rubydebug
         }
  }
}

启动logstash的命令:    .\bin\logstash.bat -f .\config\logstash.conf

4、ES:
在config目录下,elasticsearch.yml配置文件中,ES的默认端口为9200,可以通过http.port=9201来修改
启动命令:  .\bin\elasticsearch.bat
启动后,在浏览器访问:  http://localhost:9201,
如果出现了一些信息,表示启动成功。

5、Kibana:
在config目录下,kibana.yml配置文件中,通过elasticsearch.url: "http://localhost:9201" 来配置地址
启动命令:   .\bin\kibana.bat
访问  "http://localhost:9201" ,如果出现需要输入用户名密码的界面,表示ES启动失败,如果直接显示Kibana界面,则启动成功。

logstash数据输出到ES时,会选择ES默认的映射来解析数据。如果觉得默认映射不满足使用条件,可以自定模板:
借助postman工具创建ES模板,如下:
windows下如何实现kafka+ELK的日志系统
收到此响应表示创建成功。然后可以通过GET请求查询你刚才创建的模板,通过DELETE请求删除刚才创建的模板。

向kafka主题中发送日志信息,发送的信息会在cmd窗口显示,如下:
windows下如何实现kafka+ELK的日志系统
同时,在Kibana界面会看到相应的索引,索引名称就是在logstash.conf的输出中配置的,如下:
windows下如何实现kafka+ELK的日志系统
Visualize:可以选择自己想要的索引去进行图形分析,效果如下:
windows下如何实现kafka+ELK的日志系统
Dashboard:将做过的图整合到仪表盘,效果如下:
windows下如何实现kafka+ELK的日志系统

另外,可以设置自动刷新,即当有新的数据发送到kafka时,所做的图会自动根据主题来进行刷新。
项目中,可以通过logback来讲日志输出到kafka,具体配置如下:


      
        %msg%n
      
            
      bas-binding-topic
      
      
      
      bootstrap.servers=localhost:9092,10.8.22.13:9092
      
    
    
       
    

然后在记日志的时候,记到对应的主题,日志就会写到kafka相应的位置。可以通过消费者来消费主题,查看日志是否成功写入,如下:

@SuppressWarnings("resource")
  public static void main(String[] args) {

    Properties properties = new Properties();
    properties.put("bootstrap.servers", "127.0.0.1:9092");
    properties.put("group.id", "group-1");
    properties.put("enable.auto.commit", "false");
    properties.put("auto.commit.interval.ms", "1000");
    properties.put("auto.offset.reset", "earliest");
    properties.put("session.timeout.ms", "30000");
    properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put("value.deserializer", 
        "org.apache.kafka.common.serialization.StringDeserializer");

    KafkaConsumer kafkaConsumer = new KafkaConsumer<>(properties);
    kafkaConsumer.subscribe(Arrays.asList("bas-cus-request-topic", "bas-binding-topic"));
    while (true) {
      ConsumerRecords records = kafkaConsumer.poll(Long.MAX_VALUE);
      System.err.println("+++++++++++++++++++++++++++++++++++++++++");
      for (ConsumerRecord record : records) {
        System.err.println(record.offset() + ">>>>>>" + record.value());
      }
      System.err.println("+++++++++++++++++++++++++++++++++++++++++++++");
      break;
    }
  }

以上是“windows下如何实现kafka+ELK的日志系统”这篇文章的所有内容,感谢各位的阅读!希望分享的内容对大家有帮助,更多相关知识,欢迎关注创新互联行业资讯频道!

另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。


网页名称:windows下如何实现kafka+ELK的日志系统-创新互联
URL网址:http://myzitong.com/article/cssesi.html