统一日志管理方案：Spring 项目 logback 日志与 logstash 和 Elasticsearch 整合

关注

发布于: 2021 年 03 月 26 日

原创/朱季谦

最近在做一个将分布式系统的日志数据通过 logstash 传到 kafka 的功能，做完之后决定业余搭一个 ELK 日志分析系统，将 logstash 采集到的日志传给 Elasticsearch。经过一番捣鼓，也把这个过程给走通了，于是写了这篇总结，可按照以下步骤搭建 logstash 采集 spring 日志数据并传输给 Elasticsearch。

首先，logstash 是一个开源的数据收集引擎，能够同时从多个来源采集到数据，并将数据转发到想存储的“库”中。例如，logstash 可以采集数据转发存储到 Elasticsearch，也可以转发到 kafka 等消息中间件里。logstash 拥有强大的插件生态，包括基本的 http、tcp、udp、file 以及 kafa、redis 等等。这些插件，在 logstash5.x 版本以上的，是已经自带了，不需要额外安装。

在基于 ssm 的开发过程中，运行 Spring 项目时，可以在控制台上看到 log 的日志打印信息，我们可以把这些日志信息的 ERROR 日志进行监听和转发存储。

如何实现 logstash 来监听 Spring 项目的日志并将 ERROR 数据进行转发存储呢？

部署架构图：

本地验证环境：win10，Spring+Mybatis+logback.xml

可以按照以下流程来实现：

1.下载 logstash。

根据以下地址来网盘获取 logstash-5.5.2 版本的：

链接：https://pan.baidu.com/s/1h7xo65P7_O76Azt0-I-2-A

提取码：95vg

还可以官网下载：https://www.elastic.co/cn/downloads/logstash

2.安装 logstash

直接把压缩包压缩到本地某个盘里就可以了，不需要做额外操作了，5.x 以上版本的 logstash 是不需要安装其他插件，已经自带大部分插件。

3.验证是否安装成功

运行 cmd，进入 logstash 的 bin 目录下，运行指令：logstash -e 'input{stdin{}} output{stdout{}}'

运行成功的截图如下，即为安装并启动成功：

启动以后，在光标处输入：hellowrold

正常情况下，会显示以下信息，证明 logstash 可以正常使用了。

在这个过程里，涉及到几个概念，logstash 是一个管道，里面有两个 input 和 output 的必选元素，即输入与输出，之间还可以有一个可选的过滤器 filter 过滤器。input 插件从源头获取到数据，过滤器会根据条件来进行修改，最后通过 ouput 插件将数据传输，可输出给 Elasticsearch、kafka、file

等。

处理过程模型图如下：

Logstash 提供了一个 shell 脚本叫 logstash，支持以下运行参数：

执行命令： -e 执行-e 后面的参数：logstash -e 'input{stdin{}} output{stdout{}}'

执行文件: --config 或 -f 执行-f 后面的 conf 文件：logstash -f logstash.conf

输入插件：input{ … }

过滤插件：filter{ … }

输出插件：output{ … }

测试配置文件是否正确，然后退出：-t

在这篇文章里，主要用到以上这些命令，其余读者若感兴趣可以自行去研究探索。

4.配置一个文件**.conf

可以在 bin 目录或者 config 目录或者其他目录下，新建一个**.conf 文件，我选择的是 bin 目录下，新建文件 logstash.conf，截图如下：

在 logstash.conf 文件里配置信息：

 1 input { stdin { } }#该行可有可无，写来打印测试而已 2 input { 3     #开启tcp插件的监听 4     tcp { 5     #这个需要配置成本机IP，不然logstash无法启动 6     host => "127.0.0.1" 7     #端口号 8     port => 9600 9     #将日志以json格式输入10     codec => json_lines11   }12 }13 14 output {15    #输出打印16     stdout { codec => rubydebug } 17 }

复制代码

配置好，就可以先启动进行监听了，启动命令：先 cd 进到存放 logstash.conf 的目录下，我的目录在 bin 里，所以进入的是 bin 目录，执行：logstash -f logstash.conf。

5.在 spring 进行 logstash 配置的 maven 依赖引入

我在项目里用到的开源日志组件是 logback，它是 log4j 的改良，主要分为以下三个模块：

logback-classic：log4j 的一个改良版本，完整实现了 slf4j API，可以方便更换成其它日志系统如 log4j 或 JDK14 Logging。

logback-access：访问模块与 Servlet 容器集成提供通过 Http 来访问日志的功能。

logback-core：是其它两个模块的基础模块。

logback 需要在 maven 里引用到的依赖：

<dependency>    <groupId>ch.qos.logback</groupId>    <artifactId>logback-classic</artifactId>    <version>1.2.3</version></dependency><dependency>    <groupId>ch.qos.logback</groupId>    <artifactId>logback-core</artifactId>    <version>1.2.3</version></dependency><dependency>    <groupId>ch.qos.logback</groupId>    <artifactId>logback-access</artifactId>    <version>1.2.3</version></dependency><dependency>    <groupId>net.logstash.log4j</groupId>    <artifactId>jsonevent-layout</artifactId>    <version>1.6</version></dependency><dependency>    <groupId>net.logstash.logback</groupId>    <artifactId>logstash-logback-encoder</artifactId>    <version>5.0</version></dependency>

复制代码

如果引用到的 ch.qos.logback 依赖版本太低的话，可能会出现以下错误：java.lang.NoSuchMethodError: ch.qos.logback.core.util.Loader.getResourceOccurrenceCount(Ljava/lang/String;Ljava/lang/ClassLoader;)Ljava/util/Set;

可以根据项目需求来选择合适的版本，经过测试，以上的 1.2.3 版本是可以符合要求的。配置完成依赖后，就可以开始进行下一步配置。

6.在 spring 的 logback.xml 里进行 logstash 配置（省略 logback 其余无关该流程的部分）

 1 <!--开启tcp格式的logstash传输，通过TCP协议连接Logstash--> 2 <appender name="STASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender"> 3     <destination>127.0.0.1:9600</destination> 4  5     <encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder"> 6         <jsonFactoryDecorator class="net.logstash.logback.decorate.CharacterEscapesJsonFactoryDecorator"> 7             <escape> 8                 <targetCharacterCode>10</targetCharacterCode> 9                 <escapeSequence>\u2028</escapeSequence>10             </escape>11         </jsonFactoryDecorator>12         <providers>13             <pattern>14                 <pattern>15                     {16                     "timestamp":"%date{ISO8601}",17                     "user":"test",18                     "message":"[%d{yyyy-MM-dd HH:mm:ss.SSS}][%p][%t][%l{80}|%L]%m"}%n19                     }20                 </pattern>21             </pattern>22         </providers>23     </encoder>24     <keepAliveDuration>5 minutes</keepAliveDuration>25 </appender>26 27 <root level="INFO">28     <appender-ref ref="STASH"/>29 </root>

复制代码

配置说明：

encoder：配置的规范；

LoggingEventCompositeJsonEncoder：json 格式的编码器，即将日志数据转换成 json 格式；

jsonFactoryDecorator：解决中文转码的问题；

providers：json 格式提供者，对 json 进行一个定制化设置，比如，timestamp，message，thread_name 等，其他的自定义的字段的值可以通过 MDC 设置进来，格式就是 %date{xx},

注意：按照上面的设置，logstash 才可以正常接收到日志数据，否则是无法接收到的。

destination 定义的 ip 与端口与 logstash 里的 logstash.conf 需一直，logstash.conf 里的 tcp 会一直监听这个 ip 的端口：

配置完成后，启动 spring 项目，这时原来监听 tcp 的 logstash 就可以实时监听接收到了数据，logstash 的控制台显示打印如下：

若要监听到的是 ERROR 级别的日志，在 logback.xml 里的 logstash 配置里的 appender 里添加一行以下代码即可：

<filter class="ch.qos.logback.classic.filter.ThresholdFilter">    <level>ERROR</level></filter>

复制代码

在日志级别修改为以下即可：1<root level="INFO">2<appender-ref ref="STASH"/>3<appender-ref ref="ERROR"/>4</root>

7.到这一步，就完成了通过 logstash 收集 spring 的 logback 日志的功能，在这个基础上，可以再进一步扩展，扩展将 logstash 采集到的数据输出到 Elasticsearch。

 1  input { stdin { } }#该行可有可无 2  input { 3      #开启tcp模式的监听 4      tcp { 5      #这个需要配置成本机IP，不然logstash无法启动 6      host => "127.0.0.1" 7      #端口号 8      port => 9600 9      #将日志以json格式输入10      codec => json_lines11     }12 }13  14 output {15     #输出打印16     stdout { codec => rubydebug } 17     elasticsearch { hosts => ["127.0.0.1:9200"] }18 }