基於Flume+Log4j+Kafka的日誌採集架構方案（下）
爲了可以準確的捕獲到異常數據，咱們還須要對程序進行一些規範化的改造，例如提供統一的異常處理句柄等等。java
既然打算要對日誌進行統一處理，一個統1、規範的日誌格式就是很是重要的，而咱們以往使用的 PatternLayout 對於最終字段的切分很是的不方便，以下所示：linux
2016-05-08 19:32:55,572 [INFO ] [main] - [com.banksteel.log.demo.log4j.Demo.main(Demo.java:13)] 輸出信息……
2016-05-08 19:32:55,766 [DEBUG] [main] - [com.banksteel.log.demo.log4j.Demo.main(Demo.java:15)] 調試信息……
2016-05-08 19:32:55,775 [WARN ] [main] - [com.banksteel.log.demo.log4j.Demo.main(Demo.java:16)] 警告信息……
2016-05-08 19:32:55,783 [ERROR] [main] - [com.banksteel.log.demo.log4j.Demo.main(Demo.java:20)] 處理業務邏輯的時候發生一個錯誤……
java.lang.Exception: 錯誤消息啊
at com.banksteel.log.demo.log4j.Demo.main(Demo.java:18)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)apache
如何去解析這個日誌，是個很是頭疼的地方，萬一某個系統的開發人員輸出的日誌不符合既定規範的 PatternLayout 就會引起異常。json
爲了可以一勞永逸的解決格式問題，咱們採用 JsonLayout 就能很好的規範日誌輸出，例如LOG4J 2.X 版本中提供的 JsonLayout 輸出的格式以下所示：bootstrap
{
"timeMillis" : 1462712870612,
"thread" : "main",
"level" : "FATAL",
"loggerName" : "com.banksteel.log.demo.log4j2.Demo",
"message" : "發生了一個可能會影響程序繼續運行下去的異常！",
"thrown" : {
"commonElementCount" : 0,
"localizedMessage" : "錯誤消息啊",
"message" : "錯誤消息啊",
"name" : "java.lang.Exception",
"extendedStackTrace" : [ {
"class" : "com.banksteel.log.demo.log4j2.Demo",
"method" : "main",
"file" : "Demo.java",
"line" : 20,
"exact" : true,
"location" : "classes/",
"version" : "?"
}, {
"class" : "sun.reflect.NativeMethodAccessorImpl",
"method" : "invoke0",
"file" : "NativeMethodAccessorImpl.java",
"line" : -2,
"exact" : false,
"location" : "?",
"version" : "1.7.0_80"
}, {
"class" : "sun.reflect.NativeMethodAccessorImpl",
"method" : "invoke",
"file" : "NativeMethodAccessorImpl.java",
"line" : 57,
"exact" : false,
"location" : "?",
"version" : "1.7.0_80"
}, {
"class" : "sun.reflect.DelegatingMethodAccessorImpl",
"method" : "invoke",
"file" : "DelegatingMethodAccessorImpl.java",
"line" : 43,
"exact" : false,
"location" : "?",
"version" : "1.7.0_80"
}, {
"class" : "java.lang.reflect.Method",
"method" : "invoke",
"file" : "Method.java",
"line" : 606,
"exact" : false,
"location" : "?",
"version" : "1.7.0_80"
}, {
"class" : "com.intellij.rt.execution.application.AppMain",
"method" : "main",
"file" : "AppMain.java",
"line" : 144,
"exact" : true,
"location" : "idea_rt.jar",
"version" : "?"
} ]
},
"endOfBatch" : false,
"loggerFqcn" : "org.apache.logging.log4j.spi.AbstractLogger",
"source" : {
"class" : "com.banksteel.log.demo.log4j2.Demo",
"method" : "main",
"file" : "Demo.java",
"line" : 23
}
}
咱們看到，這種格式，不管用什麼語言都能輕鬆解析了。api
日誌框架的Kafka集成
咱們這裏只用log4j 1.x 和 log4j 2.x 進行示例。app
log4j 1.x 與 Kafka 集成
首先POM.xml的內容以下：框架
<dependencies>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.7.4</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.7.4</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
<version>2.7.4</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.8.2.1</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.8.2.1</version>
</dependency>
</dependencies>
注意，咱們這裏使用的Kafka版本號是0.8.2.1，可是對應0.9.0.1是可使用的而且0.9.0.1也只能用0.8.2.1纔不會發生異常（具體異常能夠本身嘗試一下）。ide
而log4j 1.x 自己是沒有 JsonLayout 可用的，所以咱們須要本身實現一個類，以下所示：學習
package com.banksteel.log.demo.log4j;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.log4j.Layout;
import org.apache.log4j.spi.LoggingEvent;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
/**
* 擴展Log4j 1.x，使其支持 JsonLayout，與 log4j2.x 同樣是基於Jackson進行解析，其格式也是徹底參考 Log4J 2.x實現的。
*
* @author 熱血BUG男
* @version 1.0.0
* @since Created by gebug on 2016/5/8.
*/
public class JsonLayout extends Layout {
private final ObjectMapper mapper = new ObjectMapper();
public String format(LoggingEvent loggingEvent) {
String json;
Map<String, Object> map = new LinkedHashMap<String, Object>(0);
Map<String, Object> source = new LinkedHashMap<String, Object>(0);
source.put("method", loggingEvent.getLocationInformation().getMethodName());
source.put("class", loggingEvent.getLocationInformation().getClassName());
source.put("file", loggingEvent.getLocationInformation().getFileName());
source.put("line", safeParse(loggingEvent.getLocationInformation().getLineNumber()));
map.put("timeMillis", loggingEvent.getTimeStamp());
map.put("thread", loggingEvent.getThreadName());
map.put("level", loggingEvent.getLevel().toString());
map.put("loggerName", loggingEvent.getLocationInformation().getClassName());
map.put("source", source);
map.put("endOfBatch", false);
map.put("loggerFqcn", loggingEvent.getFQNOfLoggerClass());
map.put("message", safeToString(loggingEvent.getMessage()));
map.put("thrown", formatThrowable(loggingEvent));
try {
json = mapper.writeValueAsString(map);
} catch (JsonProcessingException e) {
return e.getMessage();
}
return json;
}
private List<Map<String, Object>> formatThrowable(LoggingEvent le) {
if (le.getThrowableInformation() == null ||
le.getThrowableInformation().getThrowable() == null)
return null;
List<Map<String, Object>> traces = new LinkedList<Map<String, Object>>();
Map<String, Object> throwableMap = new LinkedHashMap<String, Object>(0);
StackTraceElement[] stackTraceElements = le.getThrowableInformation().getThrowable().getStackTrace();
for (StackTraceElement stackTraceElement : stackTraceElements) {
throwableMap.put("class", stackTraceElement.getClassName());
throwableMap.put("file", stackTraceElement.getFileName());
throwableMap.put("line", stackTraceElement.getLineNumber());
throwableMap.put("method", stackTraceElement.getMethodName());
throwableMap.put("location", "?");
throwableMap.put("version", "?");
traces.add(throwableMap);
}
return traces;
}
private static String safeToString(Object obj) {
if (obj == null) return null;
try {
return obj.toString();
} catch (Throwable t) {
return "Error getting message: " + t.getMessage();
}
}
private static Integer safeParse(String obj) {
try {
return Integer.parseInt(obj.toString());
} catch (NumberFormatException t) {
return null;
}
}
public boolean ignoresThrowable() {
return false;
}
public void activateOptions() {
}
}
其實並不複雜，注意其中有一些獲取不到的信息，用？代替了，保留字段的目的在於與log4j 2.x 的日誌格式徹底一致，配置log4j.properties以下對接 Kafka：
log4j.rootLogger=INFO,console log4j.logger.com.banksteel.log.demo.log4j=DEBUG,kafka log4j.appender.kafka=kafka.producer.KafkaLog4jAppender log4j.appender.kafka.topic=server_log log4j.appender.kafka.brokerList=Kafka-01:9092,Kafka-02:9092,Kafka-03:9092 log4j.appender.kafka.compressionType=none log4j.appender.kafka.syncSend=true log4j.appender.kafka.layout=com.banksteel.log.demo.log4j.JsonLayout # appender console log4j.appender.console=org.apache.log4j.ConsoleAppender log4j.appender.console.target=System.out log4j.appender.console.layout=org.apache.log4j.PatternLayout log4j.appender.console.layout.ConversionPattern=%d [%-5p] [%t] - [%l] %m%n
經過打印日誌咱們能夠看到其輸出的最終格式以下：
{
"timeMillis": 1462713132695,
"thread": "main",
"level": "ERROR",
"loggerName": "com.banksteel.log.demo.log4j.Demo",
"source": {
"method": "main",
"class": "com.banksteel.log.demo.log4j.Demo",
"file": "Demo.java",
"line": 20
},
"endOfBatch": false,
"loggerFqcn": "org.slf4j.impl.Log4jLoggerAdapter",
"message": "處理業務邏輯的時候發生一個錯誤……",
"thrown": [
{
"class": "com.intellij.rt.execution.application.AppMain",
"file": "AppMain.java",
"line": 144,
"method": "main",
"location": "?",
"version": "?"
},
{
"class": "com.intellij.rt.execution.application.AppMain",
"file": "AppMain.java",
"line": 144,
"method": "main",
"location": "?",
"version": "?"
},
{
"class": "com.intellij.rt.execution.application.AppMain",
"file": "AppMain.java",
"line": 144,
"method": "main",
"location": "?",
"version": "?"
},
{
"class": "com.intellij.rt.execution.application.AppMain",
"file": "AppMain.java",
"line": 144,
"method": "main",
"location": "?",
"version": "?"
},
{
"class": "com.intellij.rt.execution.application.AppMain",
"file": "AppMain.java",
"line": 144,
"method": "main",
"location": "?",
"version": "?"
},
{
"class": "com.intellij.rt.execution.application.AppMain",
"file": "AppMain.java",
"line": 144,
"method": "main",
"location": "?",
"version": "?"
}
]
}
測試類：
package com.banksteel.log.demo.log4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author 熱血BUG男
* @version 1.0.0
* @since Created by gebug on 2016/5/8.
*/
public class Demo {
private static final Logger logger = LoggerFactory.getLogger(Demo.class);
public static void main(String[] args) {
logger.info("輸出信息……");
logger.trace("隨意打印……");
logger.debug("調試信息……");
logger.warn("警告信息……");
try {
throw new Exception("錯誤消息啊");
} catch (Exception e) {
logger.error("處理業務邏輯的時候發生一個錯誤……", e);
}
}
}
log4j 2.x 與 Kafka 集成
log4j 2.x 天生支持 JsonLayout，而且與 Kafka 集成方便，咱們只須要循序漸進的配置一下就行了，POM.xml以下：
<dependencies>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.7.4</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.7.4</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
<version>2.7.4</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.9.0.1</version>
</dependency>
</dependencies>
log4j2.xml配置文件以下所示：
<?xml version="1.0" encoding="UTF-8"?>
<!-- Log4j2 的配置文件 -->
<Configuration status="DEBUG" strict="true" name="LOG4J2_DEMO" packages="com.banksteel.log.demo.log4j2">
<properties>
<property name="logPath">log</property>
</properties>
<Appenders>
<!--配置控制檯輸出樣式-->
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%highlight{%d{yyyy-MM-dd HH:mm:ss} %d{UNIX_MILLIS} [%t] %-5p %C{1.}:%L - %msg%n}"/>
</Console>
<!-- 配置Kafka日誌主動採集，Storm會將日誌解析成字段存放在HBase中。 -->
<Kafka name="Kafka" topic="server_log">
<!--使用JSON傳輸日誌文件-->
<JsonLayout complete="true" locationInfo="true"/>
<!--Kafka集羣配置，須要在本機配置Hosts文件，或者經過Nginx配置-->
<Property name="bootstrap.servers">Kafka-01:9092,Kafka-02:9092,Kafka-03:9092</Property>
</Kafka>
</Appenders>
<Loggers>
<Root level="DEBUG">
<!--啓用控制檯輸出日誌-->
<AppenderRef ref="Console"/>
<!--啓用Kafka採集日誌-->
<AppenderRef ref="Kafka"/>
</Root>
</Loggers>
</Configuration>
這樣就Okay了，咱們能夠在Kafka中看到完整的輸出：
{
"timeMillis" : 1462712870591,
"thread" : "main",
"level" : "ERROR",
"loggerName" : "com.banksteel.log.demo.log4j2.Demo",
"message" : "處理業務邏輯的時候發生一個錯誤……",
"thrown" : {
"commonElementCount" : 0,
"localizedMessage" : "錯誤消息啊",
"message" : "錯誤消息啊",
"name" : "java.lang.Exception",
"extendedStackTrace" : [ {
"class" : "com.banksteel.log.demo.log4j2.Demo",
"method" : "main",
"file" : "Demo.java",
"line" : 20,
"exact" : true,
"location" : "classes/",
"version" : "?"
}, {
"class" : "sun.reflect.NativeMethodAccessorImpl",
"method" : "invoke0",
"file" : "NativeMethodAccessorImpl.java",
"line" : -2,
"exact" : false,
"location" : "?",
"version" : "1.7.0_80"
}, {
"class" : "sun.reflect.NativeMethodAccessorImpl",
"method" : "invoke",
"file" : "NativeMethodAccessorImpl.java",
"line" : 57,
"exact" : false,
"location" : "?",
"version" : "1.7.0_80"
}, {
"class" : "sun.reflect.DelegatingMethodAccessorImpl",
"method" : "invoke",
"file" : "DelegatingMethodAccessorImpl.java",
"line" : 43,
"exact" : false,
"location" : "?",
"version" : "1.7.0_80"
}, {
"class" : "java.lang.reflect.Method",
"method" : "invoke",
"file" : "Method.java",
"line" : 606,
"exact" : false,
"location" : "?",
"version" : "1.7.0_80"
}, {
"class" : "com.intellij.rt.execution.application.AppMain",
"method" : "main",
"file" : "AppMain.java",
"line" : 144,
"exact" : true,
"location" : "idea_rt.jar",
"version" : "?"
} ]
},
"endOfBatch" : false,
"loggerFqcn" : "org.apache.logging.log4j.spi.AbstractLogger",
"source" : {
"class" : "com.banksteel.log.demo.log4j2.Demo",
"method" : "main",
"file" : "Demo.java",
"line" : 22
}
}
爲了減小日誌對空間的佔用，咱們一般還會設置JSONLayout的compact屬性爲true，這樣在kafka中得到的日誌將會排除掉空格和換行符。
最後
因爲在實際開發中，咱們會引入多個第三方依賴，這些依賴每每也會依賴無數的log日誌框架，爲了保證測試經過，請認清本文例子中的包名以及版本號，log4j 1.x 的 Json 輸出是爲了徹底模擬 2.x 的字段，所以部分字段用？代替，若是想要完美，請自行解決。
隨便解釋一下日誌級別，以便創建規範：
log.error 錯誤信息，一般寫在 catch 中，可使用 log.error("發生了一個錯誤",e) 來記錄詳細的異常堆棧
log.fatal 嚴重錯誤，該級別的錯誤用來記錄會致使程序異常退出的錯誤日誌。
log.warn 警告
log.info 信息
log.trace 簡單輸出文字
log.debug 調試信息
Log4j配置詳解 http://www.linuxidc.com/Linux/2014-10/108401.htm
Apache Log4j 2 更多內容請看： http://logging.apache.org/log4j/2.x/
Log4j入門使用教程 http://www.linuxidc.com/Linux/2013-06/85223.htm
Log4j 日誌詳細用法 http://www.linuxidc.com/Linux/2014-09/107303.htm
Hibernate配置Log4j顯示SQL參數 http://www.linuxidc.com/Linux/2013-03/81870.htm
Log4j學習筆記(1)_Log4j 基礎&配置項解析 http://www.linuxidc.com/Linux/2013-03/80586.htm
Log4j學習筆記(2)_Log4j配置示例&Spring集成Log4j http://www.linuxidc.com/Linux/2013-03/80587.htm
相關文章
- 1. 基於Flume+Log4j+Kafka的日誌採集架構方案
- 2. 基於Flume+Log4j+Kafka的日誌採集架構方案（上）
- 3. 基於Flink的日誌採集
- 4. 日誌採集框架Flume
- 5. Flume—日誌採集框架
- 6. Polling + Inotify 組合下的日誌保序採集方案
- 7. 數據採集 | 分佈式日誌採集-ELK方案設計
- 8. 微服務架構下日誌採集運維管理分析
- 9. Flume日誌採集框架的使用
- 10. flume的日誌採集框架
- 更多相關文章...
- • Docker 架構 - Docker教程
- • Spring基於Annotation裝配Bean - Spring教程
- • ☆基於Java Instrument的Agent實現
- • 常用的分佈式事務解決方案
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. No provider available from registry 127.0.0.1:2181 for service com.ddbuy.ser 解決方法
- 2. Qt5.7以上調用虛擬鍵盤(支持中文)，以及源碼修改(可拖動，水平縮放)
- 3. 軟件測試面試- 購物車功能測試用例設計
- 4. ElasticSearch（概念篇）：你知道的, 爲了搜索…
- 5. redux理解
- 6. gitee創建第一個項目
- 7. 支持向量機之硬間隔（一步步推導，通俗易懂）
- 8. Mysql 異步複製延遲的原因及解決方案
- 9. 如何在運行SEPM配置嚮導時將不可認的複雜數據庫密碼改爲簡單密碼
- 10. windows系統下tftp服務器使用
歡迎關注本站公眾號,獲取更多信息
