ELK 是 Elasticsearch 、 Logstash 、 Kibana 的简称

使用 spring aop 进行日志收集，然后通过 kafka 将日志发送给 logstash ， logstash 再将日志写入

elasticsearch ，这样 elasticsearch 就有了日志数据了，最后，则使用 kibana 将存放在

elasticsearch 中的日志数据显示出来，并且可以做实时的数据图表分析等等。

 

 1.docker环境搭建ELK+Kafka

# 修改为宿主机 IP 如 192.168.31.113

advertised.listeners = PLAINTEXT : //kafka : 9092

 

修改logstash.conf文件

input{

kafka {

bootstrap_servers => ["kafka:9092"] # 修改为 kafka 的 IP

auto_offset_reset => "latest" 将文件夹 kafka 与 logstash 复制到服务器，执行命令，生产镜像

ProviderController.java

consumer_threads => 5

decorate_events => true

topics => ["user-error"] # 数组形式，可以填写多个

type => "user-error" # 可以自由指定

}

}

output {

elasticsearch {

hosts => [ "elasticsearch:9200"] # 指向 Elasticsearch 服务地址，可以有多个，注意

IP 和端口和实际保持一致

index => "%{[type]}log-%{+YYYY-MM-dd}"

}

}

 将文件夹kafka与logstash复制到服务器，执行命令，生产镜像

docker build -t kgc/logstash 路径 /logstash/

根据镜像，生产容器

#kafka

docker run -d --name kgc_kafka -p 9092:9092 --network kgc_elastic_cluster --

network-alias kafka kgc/kafka

#logstash

docker run -d -it --name kgc_logstash --network kgc_elastic_cluster --network

alias logstash kgc/logstash

 

 编写程序，修改端口号与连接kafka

server :

port : 8088

spring :

kafka :

producer :

bootstrap-servers : 192.168.31.113 : 9092

ProviderController.java

@RestController
public class ProviderController {
@Autowired
private KafkaTemplate<String, String> KafkaTemplate;
@RequestMapping(value = "/test" )
public String test() throws Exception{
System.out.println(KafkaTemplate);
for (int i = 0; i < 10; i++) {
KafkaTemplate.send("wangzhuanyun", "dm", "wzy222222--->" + i);
}
return "send msg over...";
}
}