分布式日志处理：ELK+Kafka实现日志收集

发布时间：2026/11/7 2:49:40

分布式日志处理：

Sleuth实现微服务跟踪

ELK+Kafka实现日志收集系统

背景：

主要针对，分布式项目！对多个模块的日志进行管理…
在传统项目中，如果在生产环境中，有多台不同的服务器集群，
如果生产环境需要通过日志定位项目的Bug的话，需要在每台节点上使用传统的命令方式查询，这样效率非常低下。

Linux环境搭建：

不需要朋友之间跳过即可！应用开发
这里一般公司中，都会有专业的运维人员来进行搭建，个人学习就顺手学习了整理了笔记！
适合个人学习搭建

真实的环境安装也大致如此，当然本人不是专业滴，错误地方希望提醒学习！
本人VM虚拟机的,网络连接方式，采用NAT网络地址转换不细致讲解了… 如果是Bridged Adapter桥接可能会麻烦…
本人喜欢工具都安装在一个目录下, 方便查找：cd /usrl/local
别忘了关闭防火墙！

Elasticsearch

elasticsearch-6.2.4.tar.gz 安装文件
elasticsearch-analysis-ik-6.2.4.zip 安装中文文件

sudosysctl-wvm.max_map_count=262144#设置内存！每次启动都要给个运行内存！tar-zxvfelasticsearch-6.2.4.tar.gz#指定目录下 解压#解压完成,设置elasticsearch-6.2.4/config/目录下：elasticsearch.yml文件保存这一些配置信息,端口9200...#linunx 创建一个组 Elasticsearch 不能使用Root用户，需要由一个独立用户！groupaddelsearch#创建一个组useraddelsearch-gelsearch-pelasticsearch#设置密码#解压elasticsearch中文安装包！#unzip unzip 解压命令! zip压缩包解压,需要下载应用！ install 在线安装(需要网络！)！yuminstallunzip#下载unzip应用;#解压目录下文件！unzip/usr/local/elasticsearch-analysis-ik-6.2.4.zip#授权用户操作elsearchchown-Relsearch:elsearch elasticsearch-6.2.4#创建一个目录, 把中午解压包放进去。 就可以显示中文了！mkdir-p/usr/local/elasticsearch-6.2.4/plugins/analysis-ik/#elasticsearch-6.2.4/plugins下创建一个文件analysis-ik#中文解压文件,直接copy进去cp-r/usr/local/elasticsearch/* /usr/local/elasticsearch-6.2.4/plugins/analysis-ik/#Contos还需进行配置：修改 soft nproc 4096vi/etc/security/limits.d/20-nproc.conf#修改 soft nproc 4096 对应...vi/etc/security/limits.conf#文件添加：↓↓* soft nofile65536* hard nofile65536* soft nproc4096* hard nproc4096#授权：chown-Relsearch:elsearch /usr/local/elasticsearch/

启动 Elasticsearch

# 切换用户suelsearch# 刷新source/etc/profile# elasticsearch /bin目录下启动./elasticsearch#启动Elasticsearch

根据个人情况使用，不同人的配置用户…都不一样!
启动成功显示端口！

Windows 连接测试：

输入Linux 暴漏 ip 端口

注意：root 配置永久启用空间，就不用没有启动，赋值运行空间！

启动莫名奇妙报错了：
ERROR: [1] bootstrap checks failed [1]: max virtual memory areas vm.max_map_

切换到root用户修改配置sysctl.conf 执行：vi /etc/sysctl.conf
添加下面配置：vm.max_map_count=655360
并执行命令：sysctl -p
然后，重新启动elasticsearch，即可启动成功。虽然不太清除怎么回事不过，唉！好了！

Logstash

logstash-6.3.0.tar.gz

#解压tar-zxvflogstash-6.3.0.tar.gz#修改配置,logstash-6.3.0/bin/logstash.conf#修改配置,logstash-6.3.0/config/logstash.yml#启动...比较慢！#/usr/local 是我的路径,注意要根据自己的来！/usr/local/logstash-6.3.0/bin/logstash-f/usr/local/logstash-6.3.0/bin/logstash.conf

logstash.conf

Logstash从Kafka中读取消息在转发到Elasticsearch中保存。kafka与Elasticsearch 直接通信直接的传输工具！

input 读取获取目标，output写入输出目标
修改写入方kafka的 ip 端口配置，线程，主题…
输出方Elasticsearch的端口配置…

logstash.yml

logstash 对 Elasticsearch 的ip端口引用依赖！

启动：

_有点慢…

Kibana

kibana-6.2.4-linux-x86_64

#解压tar-zxvfkibana-6.2.4-linux-x86_64#配置：Elasticsearch 的ip端口引用依赖！ /usr/local/kibana-6.2.4-linux-x86_64/config/kibana.yml 文件！#更改ip#bin目录启动：#/usr/local/kibana-6.2.4-linux-x86_64/bin#./kibana#启动

kibana.yml

启动：`./kibana`

Windows查看图形化界面！默认端口5601

kafka

kafka_2.10-0.10.2.1.tgz

#root 用户下解压：tar-zxvfkafka_2.10-0.10.2.1.tgz#配置:kafka_2.10-0.10.2.1/config/server.properties文件,#内置配置：监听器,将内部的默认ip端口更改为,linux 本机的ip 0.0.0.0 指所有的ip都可以访问，但不建议更改！ 只需要把看到的IP都改了即可！

启动

kafka 启动需要两个服务！zookeeperkafka内部集成了 zookeeper

#zookeeper/usr/local/kafka_2.10-0.10.2.1/bin/zookeeper-server-start.sh-daemon/usr/local/kafka_2.10-0.10.2.1/config/zookeeper.properties#kafka/usr/local/kafka_2.10-0.10.2.1/bin/kafka-server-start.sh-daemon/usr/local/kafka_2.10-0.10.2.1/config/server.properties

测试：

这只是做测试！并不是实际应用！实际应用是程序来写读的…
bin 目录下测试

#创建 producer(生产者): 测试生产消息，产生消息, 主题：topicbin/kafka-console-producer.sh --broker-list localhost:9092-topictest#创建 consumer(消费者): 测试消费, 接受消息的 主题：topicbin/kafka-console-consumer.sh--zookeeperlocalhost:2181-topictest--from-beginning

条件允许使用 XShell 开启两个窗口，一个生产者一个消费者…开启测试：
生产者输入数据回车，消费者可以实时的监听到执行！

linux:JPS查看运行线程！

注意：

上面四个服务都要启动，日志收集器才能正常使用。

应用开发Demo：

SpringBoot 项目三板斧：

依赖：

pom.xml

<!-- 父依赖 --><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.6.RELEASE</version></parent><!-- web 和 kafka依赖！ --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 程序只需要向kafka写入日志,内部就会通过 Logstash——Elasticsearch——Kibana展示 --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

配置：

就是个普通Demo 基于默认,配置一个kafka的配置即可
application.yml

spring:kafka:producer:bootstrap-servers:192.168.1.110:9092#kafka的服务地址!key-serializer:org.apache.kafka.common.serialization.StringSerializer#消息编码方式Serializer 网络传输序列化操作！value-serializer:org.apache.kafka.common.serialization.StringSerializer

启动类：

TestRun.Java

importorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublicclassTestRun{publicstaticvoidmain(String[]args){SpringApplication.run(TestRun.class,args);}}

controller

TestController.Java

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;@RestControllerpublicclassTestController{@AutowiredprivateKafkaTemplate<String,String>kafkaTemplate;//kafkaTemplate模板对象！@GetMapping("/tolog")publicStringtolog(){//循环写日志!for(inti=0;i<10;i++){//向kafka中写入日志...日志不一定是log 只是一种记录...别死板！学习最怕死板了...// 消息主题 key 日志数据valuekafkaTemplate.send("dmservice","dmservice","kafa send message to elk--------------"+i);//消息主题与,logstash-6.3.0/bin/logstash.conf 中设置的 topics 主题匹配对应！！通过它进行匹配通信！}return"ok";}}