Kafka + ELK实现日志采集，java程序员面试宝典第五版

Kafka + ELK实现日志采集，java程序员面试宝典第五版 三、启动Zookeeper

在上面Kafka的安装教程链接中可以知道在安装Kafka的时候我们就要安装Zookeeper，因为kafka是依赖于Zookeeper的。所以在启动Kafka之前要先启动Zookeeper，如果按照教程已经配置好了系统环境变量，那么在cmd中直接输入命令：zkServer即可启动。

四、启动Kafka

通过上

《一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义》

【docs.qq.com/doc/DSmxTbFJ1cmN1R2dB】 完整内容开源分享

面教程安装kafka并修改配置后，在kafka根目录下打开cmd，输入命令：.binwindowskafka-server-start.bat .configserver.properties 即可启动kafka。

接着在kafka根目录下再打开一个cmd创建一个主题：collectionlog，用于接收日志记录：.binwindowskafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic collectionlog，主题名称可以随意，但要记住这个名称后面要用到。

五、启动Elasticsearch

进入elasticsearch的bin目录下点击elasticsearch.bat启动即可。

六、启动Logstash

需要先在Logstash的根目录下创建一个名为：core.conf 的配置文件，内容如下。该配置是kafka和ELK的关联配置，Logstash接收kafka对应的主题消息，再发往es，注意里面的topic名称要改为刚才创建的主题。

Sample Logstash configuration for creating a simple Beats -> Logstash -> Elasticsearch pipeline.

input {

kafka {

#此处是kafka的ip和端口（千万不要写成zookeeper的ip端口，否则无法收集到日志）

bootstrap_servers => “localhost:9092”

#此处是我前面提到的topic名称

topics => [“collectionlog”]

auto_offset_reset => “latest”

}

}

output {

elasticsearch {

#es的Ip和端口

hosts => [“localhost:9200”]

}

}

然后进入bin目录下打开cmd，输入命令：logstash.bat -f …core.conf 启动即可。

七、启动Kibana

进入Kibana的bin目录下点击启动kibana.bat即可。后面我们可以通过http://localhost:5601 查看到收集的日志记录。

八、logback-spring.xml配置

下面回到Spring Boot项目中，在resources目录下创建logback-spring.xml配置文件，这里面的topic名称也需要修改为刚才创建的主题，内容如下：

kafka-log-test

%red(%d{yyyy-MM-dd HH:mm:ss}) %green([%thread]) %highlight(%-5level) %boldMagenta(%logger{10}) - %cyan(%msg%n)

l o g D i r / {logDir}/ logDir/{logName}.log

${logDir}/history/test_log.%d{yyyy-MM-dd}.rar

30

%d{HH:mm:ss.SSS} %contextName [%thread] %-5level %logger{36} - %msg%n

%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n

collectionlog

bootstrap.servers=localhost:9092

acks=0

linger.ms=1000

max.block.ms=0

client.id=0

然后在application.yml中配置。

logging:

config: classpath:logback-spring.xml

九、测试接口

最后在controller中写一个测试接口，打印一条日志。

package com.hedong.controller;

import com.hedong.api.KafkaLogCollectionApi;

import lombok.extern.slf4j.Slf4j;

import org.springframework.web.bind.annotation.RestController;

@Slf4j