原創/朱季謙   最近在做一個將spring框架的專案日誌資料通過logstash傳到Elasticsearch的功能模組，也就是ELK日誌分析系統裡，經過一番搗鼓，終於把這個過程給走通了，根據自己的經驗，做了這篇總結文章，按照以下步驟可以搭建spring與logstash的集合，希望可以給各位玩logstash和Elasticsearch的讀者帶來幫助。   首先，logstash是一個開源的資料收集引擎，支援各種輸入選擇，能夠同時從多個來源採集資料，將資料轉發到想儲存的“庫”中，例如，可以轉發儲存到Elasticsearch，也可以轉發到kafka等訊息中介軟體裡。logstash擁有強大的外掛生態，包括基本的http、tcp、udp、file以及kafa、redis等等。   執行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 等。這幾個東西都是外掛，在logstash5.x版本以上的，是已經自帶了，不需要額外安裝。 處理過程模型圖如下： 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目錄等地方，我選擇的是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_lines
11   }
12 }
13 
14 output {
15    #輸出列印
16     stdout { codec => rubydebug } 
17 }

配置好，就可以先啟動進行監聽了，啟動命令：先進到存放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"}%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設定進來，格式就是%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_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接收到的資料了：