分布式日志处理Sleuth实现微服务跟踪ELKKafka实现日志收集系统背景主要针对分布式项目对多个模块的日志进行管理…在传统项目中如果在生产环境中有多台不同的服务器集群如果生产环境需要通过日志定位项目的Bug的话需要在每台节点上使用传统的命令方式查询这样效率非常低下。原理可以使用 ELK Kafka实现日志收集系统应用系统发送日志信息给Kafka然后Logstash从Kafka中读取消息在转发到Elasticsearch中保存。最后Kibana从Elasticsearch读取数据并通过web界面展示出来。Kafka相当于Logstashinput端这个除了使用Kafka还可以使用File\Redis等等。Kafka高吞吐(海量读写数据,缺点不支持对象类型传输...)、分布式消息系统概念Producer生产者(消息的来源)Consumer消费者(消息输出)Topic主题(消息传递的约定)消息系统介绍一个消息系统负责将数据从一个应用传递到另外一个应用应用只需关注于数据无需关注数据在两个或多个应用间是如何传递的。有两种主要的消息传递模式点对点传递模式、发布-订阅模式。(kafka采用)ELK: 是三个开源软件的缩写,对应着三个技术E—— Elasticsearch是个开源分布式搜索引擎提供搜集、分析、存储数据三大功能。它的特点有分布式零配置自动发现索引自动分片索引副本机制restful风格接口多数据源自动搜索负载等。L—— Logstash主要是用来日志的搜集、分析、过滤日志的工具支持大量的数据获取方式一般工作方式为c/s架构:client端安装在需要收集日志的主机上server端负责将收到的各节点日志进行过滤、修改等操作在一并发往elasticsearch上去。K—— Kibana也是一个开源和免费的工具Kibana可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面可以帮助汇总、分析和搜索重要数据日志。新增了一个FileBeat它是一个轻量级的日志收集处理工具(Agent)Filebeat占用资源少适合于在各个服务器上搜集日志后传输给Logstash官方也推荐此工具。Filebeat隶属于Beats。目前Beats包含四种工具Packetbeat搜集网络流量数据Topbeat搜集系统、进程和文件系统级别的 CPU 和内存使用情况等数据Filebeat搜集文件数据Winlogbeat搜集 Windows 事件日志数据 Logstash KibanaLinux环境搭建不需要朋友之间跳过即可应用开发这里一般公司中都会有专业的运维人员来进行搭建个人学习就顺手学习了整理了笔记适合个人学习搭建真实的环境安装也大致如此当然本人不是专业滴错误地方希望提醒学习本人VM虚拟机的,网络连接方式采用NAT网络地址转换不细致讲解了… 如果是Bridged Adapter桥接可能会麻烦…本人喜欢工具都安装在一个目录下, 方便查找cd /usrl/local别忘了关闭防火墙Elasticsearchelasticsearch-6.2.4.tar.gz 安装文件elasticsearch-analysis-ik-6.2.4.zip 安装中文文件sudosysctl-wvm.max_map_count262144#设置内存每次启动都要给个运行内存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_count655360并执行命令sysctl -p然后重新启动elasticsearch即可启动成功。虽然不太清除怎么回事不过唉好了Logstashlogstash-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.conflogstash.confLogstash从Kafka中读取消息在转发到Elasticsearch中保存。kafka与Elasticsearch 直接通信直接的传输工具input 读取获取目标output写入输出目标修改写入方kafka的 ip 端口 配置线程主题…输出方Elasticsearch的端口配置…logstash.ymllogstash 对 Elasticsearch 的ip端口引用依赖启动有点慢…Kibanakibana-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启动./kibanaWindows查看图形化界面默认端口5601kafkakafka_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查看运行线程注意上面四个服务都要启动日志收集器才能正常使用。应用开发DemoSpringBoot 项目三板斧依赖pom.xml!-- 父依赖 --parentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.1.6.RELEASE/version/parent!-- web 和 kafka依赖 --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependency!-- 程序只需要向kafka写入日志,内部就会通过 Logstash——Elasticsearch——Kibana展示 --dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId/dependency配置就是个普通Demo 基于默认,配置一个kafka的配置即可application.ymlspring: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.Javaimportorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;SpringBootApplicationpublicclassTestRun{publicstaticvoidmain(String[]args){SpringApplication.run(TestRun.class,args);}}controllerTestController.Javaimportorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;RestControllerpublicclassTestController{AutowiredprivateKafkaTemplateString,StringkafkaTemplate;//kafkaTemplate模板对象GetMapping(/tolog)publicStringtolog(){//循环写日志!for(inti0;i10;i){//向kafka中写入日志...日志不一定是log 只是一种记录...别死板学习最怕死板了...// 消息主题 key 日志数据valuekafkaTemplate.send(dmservice,dmservice,kafa send message to elk--------------i);//消息主题与,logstash-6.3.0/bin/logstash.conf 中设置的 topics 主题匹配对应通过它进行匹配通信}returnok;}}消息主题与,logstash-6.3.0/bin/logstash.conf 中设置的 topics 主题匹配对应通过它进行匹配通信测试启动程序请求查看Kibana 页面查看图形报表当然也可以直接看搜索引擎中的数据未图形化数据ip/9200/_search异常Exception thrown when sending a message with key‘dmservice’ and payloadka查看程序配置的 kafka 端口主题是否准确 
长河君 · 2026/11/6 1:11:31 · 阅读 1.2k 分布式日志处理：ELK+Kafka实现日志收集
分布式日志处理Sleuth实现微服务跟踪ELKKafka实现日志收集系统背景主要针对分布式项目对多个模块的日志进行管理…在传统项目中如果在生产环境中有多台不同的服务器集群如果生产环境需要通过日志定位项目的Bug的话需要在每台节点上使用传统的命令方式查询这样效率非常低下。原理可以使用 ELK Kafka实现日志收集系统应用系统发送日志信息给Kafka然后Logstash从Kafka中读取消息在转发到Elasticsearch中保存。最后Kibana从Elasticsearch读取数据并通过web界面展示出来。Kafka相当于Logstashinput端这个除了使用Kafka还可以使用File\Redis等等。Kafka高吞吐(海量读写数据,缺点不支持对象类型传输...)、分布式消息系统概念Producer生产者(消息的来源)Consumer消费者(消息输出)Topic主题(消息传递的约定)消息系统介绍一个消息系统负责将数据从一个应用传递到另外一个应用应用只需关注于数据无需关注数据在两个或多个应用间是如何传递的。有两种主要的消息传递模式点对点传递模式、发布-订阅模式。(kafka采用)ELK: 是三个开源软件的缩写,对应着三个技术E—— Elasticsearch是个开源分布式搜索引擎提供搜集、分析、存储数据三大功能。它的特点有分布式零配置自动发现索引自动分片索引副本机制restful风格接口多数据源自动搜索负载等。L—— Logstash主要是用来日志的搜集、分析、过滤日志的工具支持大量的数据获取方式一般工作方式为c/s架构:client端安装在需要收集日志的主机上server端负责将收到的各节点日志进行过滤、修改等操作在一并发往elasticsearch上去。K—— Kibana也是一个开源和免费的工具Kibana可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面可以帮助汇总、分析和搜索重要数据日志。新增了一个FileBeat它是一个轻量级的日志收集处理工具(Agent)Filebeat占用资源少适合于在各个服务器上搜集日志后传输给Logstash官方也推荐此工具。Filebeat隶属于Beats。目前Beats包含四种工具Packetbeat搜集网络流量数据Topbeat搜集系统、进程和文件系统级别的 CPU 和内存使用情况等数据Filebeat搜集文件数据Winlogbeat搜集 Windows 事件日志数据 Logstash KibanaLinux环境搭建不需要朋友之间跳过即可应用开发这里一般公司中都会有专业的运维人员来进行搭建个人学习就顺手学习了整理了笔记适合个人学习搭建真实的环境安装也大致如此当然本人不是专业滴错误地方希望提醒学习本人VM虚拟机的,网络连接方式采用NAT网络地址转换不细致讲解了… 如果是Bridged Adapter桥接可能会麻烦…本人喜欢工具都安装在一个目录下, 方便查找cd /usrl/local别忘了关闭防火墙Elasticsearchelasticsearch-6.2.4.tar.gz 安装文件elasticsearch-analysis-ik-6.2.4.zip 安装中文文件sudosysctl-wvm.max_map_count262144#设置内存每次启动都要给个运行内存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_count655360并执行命令sysctl -p然后重新启动elasticsearch即可启动成功。虽然不太清除怎么回事不过唉好了Logstashlogstash-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.conflogstash.confLogstash从Kafka中读取消息在转发到Elasticsearch中保存。kafka与Elasticsearch 直接通信直接的传输工具input 读取获取目标output写入输出目标修改写入方kafka的 ip 端口 配置线程主题…输出方Elasticsearch的端口配置…logstash.ymllogstash 对 Elasticsearch 的ip端口引用依赖启动有点慢…Kibanakibana-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启动./kibanaWindows查看图形化界面默认端口5601kafkakafka_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查看运行线程注意上面四个服务都要启动日志收集器才能正常使用。应用开发DemoSpringBoot 项目三板斧依赖pom.xml!-- 父依赖 --parentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.1.6.RELEASE/version/parent!-- web 和 kafka依赖 --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependency!-- 程序只需要向kafka写入日志,内部就会通过 Logstash——Elasticsearch——Kibana展示 --dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId/dependency配置就是个普通Demo 基于默认,配置一个kafka的配置即可application.ymlspring: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.Javaimportorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;SpringBootApplicationpublicclassTestRun{publicstaticvoidmain(String[]args){SpringApplication.run(TestRun.class,args);}}controllerTestController.Javaimportorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;RestControllerpublicclassTestController{AutowiredprivateKafkaTemplateString,StringkafkaTemplate;//kafkaTemplate模板对象GetMapping(/tolog)publicStringtolog(){//循环写日志!for(inti0;i10;i){//向kafka中写入日志...日志不一定是log 只是一种记录...别死板学习最怕死板了...// 消息主题 key 日志数据valuekafkaTemplate.send(dmservice,dmservice,kafa send message to elk--------------i);//消息主题与,logstash-6.3.0/bin/logstash.conf 中设置的 topics 主题匹配对应通过它进行匹配通信}returnok;}}消息主题与,logstash-6.3.0/bin/logstash.conf 中设置的 topics 主题匹配对应通过它进行匹配通信测试启动程序请求查看Kibana 页面查看图形报表当然也可以直接看搜索引擎中的数据未图形化数据ip/9200/_search异常Exception thrown when sending a message with key‘dmservice’ and payloadka查看程序配置的 kafka 端口主题是否准确