原创/朱季谦
最近在做一个将分布式系统的日志数据通过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数据进行转发存储呢？
部署架构图：
可以按照以下流程来实现：
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文件里配置信息：
代码语言：javascript复制 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_lines
11 }
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里引用到的依赖：
代码语言：javascript复制<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其余无关该流程的部分）
代码语言：javascript复制 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":"�te{ISO8601}",
17 "user":"test",
18 "message":"[%d{yyyy-MM-dd HH:mm:ss.SSS}][%p][%t][%l{80}|%L]%m"}%n
19 }
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设置进来，格式就是�te{xx},
注意：按照上面的设置，logstash才可以正常接收到日志数据，否则是无法接收到的。
destination定义的ip与端口与logstash里的logstash.conf需一直，logstash.conf里的tcp会一直监听这个ip的端口：
配置完成后，启动spring项目，这时原来监听tcp的logstash就可以实时监听接收到了数据，logstash的控制台显示打印如下：
若要监听到的是ERROR级别的日志，在logback.xml里的logstash配置里的appender里添加一行以下代码即可：
代码语言：javascript复制<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。
代码语言：javascript复制 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_lines
11 }
12 }
13
14 output {
15 #输出打印
16 stdout { codec => rubydebug }
17 elasticsearch { hosts => ["127.0.0.1:9200"] }
18 }打开已经本地安装的Elasticsearch：http://127.0.0.1:9100/，可以看到，ES可以接受到logstash接收到的数据了：
