系列课程 ElasticSearch 之第 9 篇 —— ELK （ElasticSearch、Logstash、Kibana）分布式日志收集和查看（完结）

传统日志搜索有什么问题？

如果你在软件公司里上班，有客户反馈系统出bug，要你排查日志，你们一般怎么排查？传统的排查日志的办法，就是登录到每台部署微服务的机器上，然后一般使用下面的命令行：

根据关键字查询200行：

tail -200 service.log | grep '查询的关键字'

或者：

grep -C 10 -i "查询的关键字" service.log

或者实时查询最新的100行

tail -100f service.log

如果部署的微服务在多台机器，A台机器无法搜索到，又跑去B台机器上进行搜索。这样排查日志的效率是非常低的。如果部署有十几台机器，那不累垮了？

所以，我们需要一个集中化的日志管理工具，同时还需要对日志进行检索和统计。

ELK 分布式日志收集的介绍

ElasticSearch 是一个基于 Lucene 的开源分布式搜索服务器，它的特点有：分布式，零配置，自动发现，索引自动分片，索引副本机制，restful 风格接口，多数据源，自动搜索负载等。它支持全文搜索，能够达到实时搜索。ElasticSearch 具有稳定，可靠，快速，安装使用方便等特点。

Logstash （读作 log [stæʃ] ）是一个完全开源的工具，它可以对你的日志进行收集、过滤、分析，支持大量的数据获取方法，并将其存储供以后使用（如搜索）。一般工作方式为 c/s 架构，client 端安装在需要收集日志的主机上，server 端负责将收到的各节点日志进行过滤、修改等操作，然后一起发往 ElasticSearch 上去。Logstash 事件处理有三个阶段：inputs → filters → outputs。是一个接收，处理，转发日志的工具。支持系统日志，webserver日志，错误日志，应用日志，总之包括所有可以抛出来的日志类型。

Kibana 是一个基于浏览器页面的 ElasticSearch 前端展示工具，也是一个开源和免费的工具，Kibana 可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助您汇总、分析和搜索重要数据日志。

ELK 分布式日志收集原理

1、集群中的每台微服务的节点都安装 Logstash 日志收集系统插件。

2、每台服务器节点将日志输入到 Logstash 中。

3、Logstash 将该日志格式化为 json 格式，根据每天创建不同的索引，输出到 ElasticSearch 中。

4、浏览器使用安装 Kibana 查询日志信息。

ELK 环境搭建（Windows环境）

前面的博客已经讲解了 Windows 环境如何安装 ElasticSearch 和 Kibana，还安装了 Logstash。博客地址：https://blog.csdn.net/BiandanLoveyou/article/details/115742897

既然 Logstash 要收集日志，首先就要知道日志的路径。我们以当前 SpringBoot 的项目的日志为基础，让 Logstash 收集。因此，我们需要改造一下当前项目：

先下载之前的代码。代码地址：https://pan.baidu.com/s/13QyQvO3mixrBL3Ubxh2RLA 提取码：fwfw

修改 boostrap.yml 配置如下：需要说明的是，日志文件路径的配置，可以根据自己的实际情况进行配置。

spring:application:name: ElasticSearchTestdata:elasticsearch:# 集群名称，默认是 elasticsearch，如果需要改名字，需修改配置 ES 的配置文件 elasticsearch.ymlcluster-name: elasticsearch# ElasticSearch 节点地址，注意端口号是 9300，而9200是http协议端口cluster-nodes: 127.0.0.1:9300server:port: 8080# 将SpringBoot项目作为单实例部署调试时，不需要注册到注册中心
eureka:client:fetch-registry: falseregister-with-eureka: false# 日志文件路径配置
log:file:path: D:/work/Java/IDEA/workspace/logs

pom.xml 增加 log4j 的日志依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.study</groupId><artifactId>ElasticSearchTest</artifactId><version>1.0-SNAPSHOT</version><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.0.RELEASE</version></parent><dependencies><!--Spring boot 集成包--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><!--web支持--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId></dependency><!-- 引入 Google 的集合工具包 --><dependency><groupId>com.google.collections</groupId><artifactId>google-collections</artifactId><version>1.0</version></dependency><!-- LOGGING begin --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>${slf4j.version}</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>${slf4j.version}</version></dependency><!-- common-logging 实际调用slf4j --><dependency><groupId>org.slf4j</groupId><artifactId>jcl-over-slf4j</artifactId><version>${slf4j.version}</version></dependency><!-- java.util.logging 实际调用slf4j --><dependency><groupId>org.slf4j</groupId><artifactId>jul-to-slf4j</artifactId><version>${slf4j.version}</version></dependency><!-- LOGGING end --></dependencies><dependencyManagement><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>Greenwich.SR2</version><type>pom</type><scope>import</scope></dependency></dependencies></dependencyManagement></project>

然后增加一个 logback.xml 文件，放在 source 目录下，如图：

logback.xml 具体内容如下：

<?xml version="1.0" encoding="UTF-8"?>
<configuration><!-- 属性文件 --><springProperty name="appName" source="spring.application.name" defaultValue = "server"/><springProperty name="logPath" source="log.file.path" defaultValue="/data/projects/logs/"/><!-- 默认的控制台日志输出 --><appender name="stdout" class="ch.qos.logback.core.ConsoleAppender"><layout class="ch.qos.logback.classic.PatternLayout"><pattern>%date [%thread] %-5level %logger{50}:%L - %msg%n</pattern></layout></appender><!-- 配置文件轮转 --><appender name="logfile" class="ch.qos.logback.core.rolling.RollingFileAppender"><File>${logPath}/${appName}.log</File><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><FileNamePattern>${logPath}/history/server.%d{yyyy-MM-dd}.log.gz</FileNamePattern><maxHistory>30</maxHistory></rollingPolicy><layout class="ch.qos.logback.classic.PatternLayout"><pattern>%date [%thread] %-5level %logger{50}:%L - %msg%n</pattern></layout></appender><!-- root 默认日志配置--><root level="INFO"><appender-ref ref="logfile"/><appender-ref ref="stdout"/></root></configuration>

注意，我们使用默认的文件名 logback.xml 即可，因为 log4j 启动的时候会去寻找这些默认的文件名的：

这时候，我们启动项目，就会在日志文件目录下，生成日志文件，比如我的是 D:\work\Java\IDEA\workspace\logs

OK，有了日志文件，我们现在来整合 Logstash。

我们的思路是：启动 Logstash 服务，同时指定它读取哪个日志文件，即 input；然后 Logstash 经过处理成标准的 json 格式，再输出到 ElasticSearch 节点上。

我们打开 Logstash 的 config 目录，看到一个文件：

用记事本打开，内容如下：

这就是最简单的 Logstash 配置文件，它这里没有展示 filter 过滤的内容。我们仿造它创建一个我们自己的配置文件（可以复制它的，然后重命名）：my-logstash.conf，内容如下：

# 读取的源数据：需要指定文件的路径，已经格式，type等信息
input {# 从文件读取日志信息 输送到控制台file {# 文件路径path => "D:/work/Java/IDEA/workspace/logs/ElasticSearchTest.log"# 以 json 格式读取文件codec => "json" ## 以JSON格式读取日志# 指定类型为 elasticsearchtype => "elasticsearch"# 读取的位置：从开始的位置读取start_position => "beginning"}
}# 过滤器，目前暂时没用到
# filter {
#
# }# 输出
output {# 标准输出 # stdout {}# 输出进行格式化，采用Ruby库来解析日志   stdout { codec => rubydebug }# ES 的地址（9200是http协议端口）以及索引的名字（每天一个日志）elasticsearch {hosts => ["127.0.0.1:9200"]index => "es-%{+YYYY.MM.dd}"}
}

然后，我们来测试一下。

正确的启动方式是：先启动 ElasticSearch 服务器，再启动 Logstash，最后启动微服务。这样 Logstash 就不会因为找不到 ES 服务而报错。

启动 ElasticSearch 服务器就很容易了，现在说一下如何在 windows 环境下启动 Logstash。

windows 下启动 Logstash 服务需要 CMD 命令行。步骤如下：

1、打开 CMD，进入到 Logstash 的 bin 目录下：

首先进入你本机安装 Logstash 的磁盘，比如我的是D盘。那就在 CMD 命令行上输入 D: 然后回车，效果如下：

2、然后 cd 进入到你安装 Logstash 的 bin 目录，在 CMD 命令行可以结合电脑键盘的 Tab 键，能智能识别输入，注意，每进入一层文件夹，都是使用斜杠 \ ，否则无法识别，比如我的：

D:\>cd work\Java\ElasticSearch\logstash\logstash\bin

3、启动 Logstash 服务，命令行如下：在 bin 目录下，启动 logstash，加个 -f，然后使用我们的自定义配置文件，因为 bin 目录和 config 目录在同一级，在 bin 目录里需要返回到上一级（使用 ..\），才能找到我们的自定义配置文件。第一次启动需要等待比较久（30秒吧）

D:\work\Java\ElasticSearch\logstash\logstash\bin>logstash -f ..\config\my-logstash.conf

4、这时候，我们重启我们的微服务，就能看到 Logstash 的控制台实时输出我们的日志了：

OK，我们使用 Kibana 管理后台查看：

我们在 Dev Tools 菜单下，输入 GET /es 就会自动识别 Logstash 创建的索引了。这个索引是每天都生成一个。

比如我们输入查询条件：

GET /es-2021.04.21/_search

接下来，我们可以在微服务里增加一些操作 API 接口并输出一些日志，我们首先在 BlogEntity 实体类增加 toString方法，如图（快捷键 Alt+Insert 调出小窗口，找到 toString 方法）：

然后修改 controller 类：主要是增加一些日志信息：

完整代码如下：

package com.study.controller;import com.google.common.collect.Lists;
import com.study.dao.BlogDao;
import com.study.entity.BlogEntity;
import jdk.nashorn.internal.parser.JSONParser;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.web.PageableDefault;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;import java.util.*;/*** @author biandan* @description* @signature 让天下没有难写的代码* @create 2021-04-16 下午 11:33*/
@RestController
@RequestMapping(value = "/blog")
public class BlogController {private final static Logger LOGGER = LoggerFactory.getLogger(BlogController.class);@Autowiredprivate BlogDao blogDao;/*** 根据ID查询博客信息** @param id* @return*/@RequestMapping(value = "/findById", method = RequestMethod.GET)public Optional<BlogEntity> findById(String id) {LOGGER.info("findById id=" + id);Optional<BlogEntity> blogEntity = blogDao.findById(id);LOGGER.info("blogEntity=" + blogEntity.toString());try {int k = 9 / 0;} catch (Exception e) {LOGGER.error("查询异常：", e.getMessage(), e);}return blogEntity;}/*** 根据条件查询所有博客列表** @param blogName   博客名称* @param blogDesc   博客描述* @param clickCount 点击数* @return*/@RequestMapping(value = "/findAll", method = RequestMethod.POST)public List<BlogEntity> findAll(String blogName, String blogDesc, Integer clickCount) {BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();//模糊查询博客名称if (StringUtils.isNotBlank(blogName)) {MatchQueryBuilder queryBuilder = QueryBuilders.matchQuery("blogName", blogName);boolQueryBuilder.must(queryBuilder);}//模糊查询博客描述if (StringUtils.isNotBlank(blogDesc)) {MatchQueryBuilder queryBuilder = QueryBuilders.matchQuery("blogDesc", blogDesc);boolQueryBuilder.must(queryBuilder);}//根据点击数精确查询if (null != clickCount) {TermQueryBuilder queryBuilder = QueryBuilders.termQuery("clickCount", clickCount);boolQueryBuilder.must(queryBuilder);}Iterable<BlogEntity> entities = blogDao.search(boolQueryBuilder);//使用Google的工具包List<BlogEntity> resultList = Lists.newArrayList(entities);LOGGER.info("resultList =" + resultList.toString());return resultList;}/*** 分页查询** @param blogName   博客名称* @param blogDesc   博客描述* @param clickCount 点击数量* @param pageable   分页信息，我们可以设置默认值： page、value（每页查询数量）的值* @return*/@RequestMapping(value = "/findByPage", method = RequestMethod.POST)public Page<BlogEntity> findByPage(String blogName, String blogDesc, Integer clickCount, @PageableDefault(page = 0, value = 10) Pageable pageable) {BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();//模糊查询博客名称if (StringUtils.isNotBlank(blogName)) {MatchQueryBuilder queryBuilder = QueryBuilders.matchQuery("blogName", blogName);boolQueryBuilder.must(queryBuilder);}//模糊查询博客描述if (StringUtils.isNotBlank(blogDesc)) {MatchQueryBuilder queryBuilder = QueryBuilders.matchQuery("blogDesc", blogDesc);boolQueryBuilder.must(queryBuilder);}//根据点击数精确查询if (null != clickCount) {TermQueryBuilder queryBuilder = QueryBuilders.termQuery("clickCount", clickCount);boolQueryBuilder.must(queryBuilder);}Page<BlogEntity> search = blogDao.search(boolQueryBuilder, pageable);return search;}
}

重启微服务，我们使用 postman 测试根据 ID 查询博客的接口：

这时候，我们去 Logstash 的控制台，往上翻，可以看到这行日志：

我们去 Kibana 管理后台验证：

GET /es-2021.04.22/_search
{"query": {"match": {"message": "数据库"}}}

大部分使用 Kibana 的 Dev Tools 菜单，我们也可以使用 Discover，如图：

演示的代码地址：https://pan.baidu.com/s/1l4sCiR-VPQDuzvzzbBOJPg 提取码：1e3y