看了很多ELK日子分析系统的实现，多数是写怎么搭建，至于怎么在不影响主项目的情况下、异步去记录日志没有过多说明，为此我以Logback为列，通过Logback底层原理，在不影响主代码的情况下，实现ELK日志分析系统。
PS:至于安装ELK，和ELK的说明的，可以参考一下连接：
https://www.cnblogs.com/kevingrace/p/5919021.html
https://www.cnblogs.com/yuhuLin/p/7018858.html
https://my.oschina.net/itblog/blog/547250
Logback:
Logback是由log4j创始人设计的另一个开源日志组件,官方网站： http://logback.qos.ch。
看看（最）简单的Logback配置logback.xml：
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<!-- encoder 默认配置为PatternLayoutEncoder -->
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<root level="INFO">
<appender-ref ref="STDOUT" />
</root>
</configuration>
我们配置一个ConsoleAppender就可以在控制台上打印出日志。
实现思路：
Logback（我用的是logback-classic-1.2.3版本）有个DBAppender，功能就是将日志插入数据库，这样的话我们可以仿照它，实现将日志插入到elasticsearch的功能。
我们看看DBAppender所需要加的配置：
<appender name="db-classic-mysql" class="ch.qos.logback.classic.db.DBAppender">
<connectionSource class="ch.qos.logback.core.db.DataSourceConnectionSource">
<dataSource class="com.mchange.v2.c3p0.ComboPooledDataSource">
<driverClass>com.mysql.jdbc.Driver</driverClass>
<jdbcUrl>jdbc:mysql://{$server ip}:3306/{$dbname}</jdbcUrl>
<user>{$user}</user>
<password>{$password}</password>
</dataSource>
</connectionSource>
</appender>
从配置上看就是把参数传到DBAppender里面，我们ES的Appender只需仿照它的做法就可以。
目标：找出DBAppender的sql在哪里commit
我们再看看DBAppender：
image.png
其初始化参数：
protected String insertPropertiesSQL;
protected String insertExceptionSQL;
protected String insertSQL;
protected static final Method GET_GENERATED_KEYS_METHOD;
private DBNameResolver dbNameResolver;
static final int TIMESTMP_INDEX = 1;
static final int FORMATTED_MESSAGE_INDEX = 2;
static final int LOGGER_NAME_INDEX = 3;
static final int LEVEL_STRING_INDEX = 4;
static final int THREAD_NAME_INDEX = 5;
static final int REFERENCE_FLAG_INDEX = 6;
static final int ARG0_INDEX = 7;
static final int ARG1_INDEX = 8;
static final int ARG2_INDEX = 9;
static final int ARG3_INDEX = 10;
static final int CALLER_FILENAME_INDEX = 11;
static final int CALLER_CLASS_INDEX = 12;
static final int CALLER_METHOD_INDEX = 13;
static final int CALLER_LINE_INDEX = 14;
static final int EVENT_ID_INDEX = 15;
static final StackTraceElement EMPTY_CALLER_DATA = CallerData.naInstance();
涉及sql的就只有
insertPropertiesSQL，insertExceptionSQL，insertSQL。而我们顺藤摸瓜，发现最后他们都做这样的操作：
super.start();
我们看看他的类继承关系图（idea自带这个插件，eclipse可以安装类似的插件）：
image.png
看看DBAppender继承的DBAppenderBase：
protected abstract的就有getGeneratedKeysMethod();getInsertSQL();subAppend(E var1, Connection var2, PreparedStatement var3)；secondarySubAppend(E var1, Connection var2, long var3)；
就说我们仿照DBAppender继承DBAppenderBase就要实现上面4个方法；
看看DBAppenderBase继承的UnsynchronizedAppenderBase：
protected abstract的就只有有append(E var1);
再看回DBAppenderBase的append(E eventObject)：
connection.commit();
！！！！！！！！！！！！！！！！！！！！！！！！！！！！！
！！！！！！！！！！！！就是在这里！！！！！！！！！！！！
！！！！！！！！！！！！！！！！！！！！！！！！！！！！！
这样的话我们仿照DBAppenderBase写一个ESAppender，继承UnsynchronizedAppenderBase，实现append方法，将日志插入ES里！
下面就是贴代码、贴图片时间：
项目结构图：
image.png
主类ESAppender：
package com.elk.log.appender;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.classic.spi.IThrowableProxy;
import ch.qos.logback.classic.spi.ThrowableProxyUtil;
import ch.qos.logback.core.UnsynchronizedAppenderBase;
import com.alibaba.fastjson.JSON;
import com.elk.log.utils.InitES;
import com.elk.log.vo.EsLogVo;
import com.elk.log.vo.Location;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestResult;
import io.searchbox.core.Index;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.Properties;
public class ESAppender extends UnsynchronizedAppenderBase<ILoggingEvent> {
private static JestClient jestClient ;
//索引名称
String esIndex = "java-log-#date#";
//索引类型
String esType = "java-log";
//是否打印行号
boolean isLocationInfo = true;
//运行环境
String env = "";
//es地址
String esAddress = "";
public String getEsIndex() {
return esIndex;
}
public void setEsIndex(String esIndex) {
this.esIndex = esIndex;
}
public String getEsType() {
return esType;
}
public void setEsType(String esType) {
this.esType = esType;
}
public boolean isLocationInfo() {
return isLocationInfo;
}
public void setLocationInfo(boolean locationInfo) {
isLocationInfo = locationInfo;
}
public String getEnv() {
return env;
}
public void setEnv(String env) {
this.env = env;
}
public String getEsAddress() {
return esAddress;
}
public void setEsAddress(String esAddress) {
this.esAddress = esAddress;
}
@Override
protected void append(ILoggingEvent event) {
EsLogVo esLogVo = new EsLogVo();
esLogVo.setHost("HostName");
esLogVo.setIp("127.0.0.1");
esLogVo.setEnv(this.env);
esLogVo.setLevel(event.getLevel().toString());
Location location = new Location();
StackTraceElement[] callerDataArray = event.getCallerData();
if(callerDataArray != null && callerDataArray.length >0){
StackTraceElement immediateCallerData = callerDataArray[0];
location.setClassName(immediateCallerData.getClassName());
location.setMethod(immediateCallerData.getMethodName());
location.setFile(immediateCallerData.getFileName());
location.setLine(Integer.toString(immediateCallerData.getLineNumber()));
}
IThrowableProxy tp = event.getThrowableProxy();
if (tp != null){
String throwable = ThrowableProxyUtil.asString(tp);
esLogVo.setThrowable(throwable);
}
esLogVo.setLocation(location);
esLogVo.setLogger(event.getLoggerName());
esLogVo.setMessage(event.getFormattedMessage());
SimpleDateFormat df = new SimpleDateFormat("yyyy.MM.dd");
SimpleDateFormat df2 = new SimpleDateFormat("yyyy-MM-dd");
esLogVo.setTimestamp(df2.format(new Date(event.getTimeStamp())));
esLogVo.setThread(event.getThreadName());
Map<String ,String > mdcPropertyMap = event.getMDCPropertyMap();
esLogVo.setTraceId(mdcPropertyMap.get("traceId"));
esLogVo.setRpcId(mdcPropertyMap.get("rpcId"));
String jsonString = JSON.toJSONString(esLogVo);
String esIndex_format = esIndex.replace("#date#",df.format(new Date(event.getTimeStamp())));
Index index = new Index.Builder(esLogVo).index(esIndex_format).type(esType).build();
try{
JestResult result = jestClient.execute(index);
System.out.println(result);
}catch (Exception e){
e.printStackTrace();
}
}
@Override
public void start() {
super.start();
Properties properties = new Properties();
properties.put("es.hosts",esAddress);
properties.put("es.username","zkpk");
properties.put("es.password","123456");
jestClient = InitES.jestClient(properties);
}
@Override
public void stop() {
super.stop();
jestClient.shutdownClient();
}
}
我们主要实现append方法，重写start()，stop()。
append主要是插入es索引，start和stop主要用来打开和关闭ES的链接。
工具类InitES（用于初始化es连接用的）：
package com.elk.log.utils;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.config.HttpClientConfig;
import io.searchbox.client.JestClient;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Properties;
public class InitES {
private static io.searchbox.client.JestClient JestClient;
public static JestClient jestClient(Properties properties) {
JestClientFactory factory = new JestClientFactory();
String userName = properties.getProperty("es.username");
String password = properties.getProperty("es.password");
String esHosts = properties.getProperty("es.hosts");
List<String> serverList = new ArrayList <String>();
for (String address:esHosts.split(",")) {
serverList.add("http://"+address);
}
HttpClientConfig.Builder builder = new HttpClientConfig.Builder(serverList);
builder.maxTotalConnection(20);
builder.defaultMaxTotalConnectionPerRoute(5);
builder.defaultCredentials(userName,password);
builder.multiThreaded(true);
factory.setHttpClientConfig(builder.build());
if (JestClient == null) {
JestClient = factory.getObject();
}
return JestClient;
}
}
这里，我们用Jest方式连接ES。
两个Javabean：EsLogVo，Location：
public class EsLogVo {
private String host;
private String ip;
private String env;
private String message;
private String timestamp;
private String logger;
private String level;
private String thread;
private String throwable;
private Location location;
private String traceId;
private String rpcId;
...set and get...
}
public class Location {
private String className;
private String method;
private String file;
private String line;
...set and get...
}
测试类ESLogTest：
package com.elk.log;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
public class ESLogTest {
private static final Logger logger = LoggerFactory.getLogger(ESLogTest.class);
@Test
public void testLog() throws InterruptedException{
logger.info("我是正常信息 test message info");
logger.error("我一条异常的信息",new Exception("项目报错了，加班吧！！！！"));
logger.debug("debug消息 debug hello hi");
logger.warn("警告警告");
TimeUnit.SECONDS.sleep(10);
}
}
配置文件logback.xml：
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<contextName>logback-api</contextName>
<property name="logback.system" value="logback-system"/>
<property name="logback.path" value="../logs/logback-system"/>
<property name="logback.level" value="DEBUG"/>
<property name="logback.pattern" value="%d{yyyy-MM-dd HH:mm:ss} [%t] %-5p - %m%n"/>
<property name="logback.env" value="dev"/>
<property name="logback.isLocation" value="true"/>
<property name="logback.esAddress" value="192.168.0.128:9200,192.168.0.129:9200,192.168.0.130:9200"/>
<appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>[%-5p %d{yy-MM-dd HH:mm:ss} %m] %caller{1}</pattern>
</encoder>
</appender>
<appender name="ES" class="com.elk.log.appender.ESAppender">
<!--索引名字 date为通配符 ，会自动替换为yyyy.MM.dd格式-->
<esIndex>java-log-#date#</esIndex>
<!--索引类型-->
<esType>${logback.system}</esType>
<!--运行环境-->
<env>${logback.env}</env>
<!--ES地址-->
<esAddress>${logback.esAddress}</esAddress>
</appender>
<appender name="ASYNC_ES" class="ch.qos.logback.classic.AsyncAppender">
<!--默认情况下，当BlockingQueue还有20%容量，他将丢弃TRACE,DEBUG和INFO级别的event，只保留warn和error-->
<discardingThreshold>0</discardingThreshold>
<!--BlockingQueue的最大容量，默认情况下，大小为256-->
<queueSize>256</queueSize>
<appender-ref ref="ES"/>
<!--要是保留行号，需要开启为true-->
<includeCallerData>true</includeCallerData>
</appender>
<logger name="com.elk.log" additivity="true">
<level value="${logback.level}"></level>
<appender-ref ref="ASYNC_ES" />
</logger>
<root level="${logback.level}">
<appender-ref ref="stdout" />
</root>
</configuration>
这里我们定义自己的ESAppender，取名未ES，并将所需参数传到ESAppender。
然后我们再把它放到AsyncAppender里面进行异步处理，防止ELK报错影响到主程序。
OK,测试：
我们启动ELK，在Kibana的Dev Tools上创建一个java的动态日志模板：
PUT _template/java-log
{
"template" : "java-log-*",
"order" : 0,
"settings" : {
"index":{
"refresh_interval": "5s"
}
},
"mappings": {
"_default_": {
"dynamic_templates": [
{
"message_field": {
"match_mapping_type": "string",
"path_match": "message",
"mapping": {
"norms": false,
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_max_word"
}
}
},
{
"throwable_fields": {
"match_mapping_type": "string",
"path_match": "throwable",
"mapping": {
"norms": false,
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_max_word"
}
}
},
{
"string_fields": {
"match_mapping_type": "string",
"match": "*",
"mapping": {
"norms": false,
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_max_word",
"fields": {
"keyword": {
"type": "keyword"
}
}
}
}
}
],
"_all": {
"enabled": false
},
"properties": {
"env": {
"type": "keyword"
},
"host": {
"type": "keyword"
},
"ip": {
"type": "ip"
},
"level": {
"type": "keyword"
},
"location": {
"properties": {
"line": {
"type": "integer"
}
}
},
"timestamp": {
"type": "date"
}
}
}
}
}
用的是IK分词器，refresh时间定为5s用于生产上调高效率，定义了几个参数作为关键搜索。
然后将java-log-*作为主要观察对象，回到Discover
回到idea，在ESLogTest上跑testLog：
[INFO 18-03-07 16:56:49 我是正常信息 test message info] Caller+0 at com.elk.log.ESLogTest.testLog(ESLogTest.java:14)
[ERROR 18-03-07 16:56:49 我一条异常的信息] Caller+0 at com.elk.log.ESLogTest.testLog(ESLogTest.java:15)
java.lang.Exception: 项目报错了，加班吧！！！！
at com.elk.log.ESLogTest.testLog(ESLogTest.java:15)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.junit.internal.runners.TestMethod.invoke(TestMethod.java:59)
at org.junit.internal.runners.MethodRoadie.runTestMethod(MethodRoadie.java:98)
at org.junit.internal.runners.MethodRoadie$2.run(MethodRoadie.java:79)
at org.junit.internal.runners.MethodRoadie.runBeforesThenTestThenAfters(MethodRoadie.java:87)
at org.junit.internal.runners.MethodRoadie.runTest(MethodRoadie.java:77)
at org.junit.internal.runners.MethodRoadie.run(MethodRoadie.java:42)
at org.junit.internal.runners.JUnit4ClassRunner.invokeTestMethod(JUnit4ClassRunner.java:88)
at org.junit.internal.runners.JUnit4ClassRunner.runMethods(JUnit4ClassRunner.java:51)
at org.junit.internal.runners.JUnit4ClassRunner$1.run(JUnit4ClassRunner.java:44)
at org.junit.internal.runners.ClassRoadie.runUnprotected(ClassRoadie.java:27)
at org.junit.internal.runners.ClassRoadie.runProtected(ClassRoadie.java:37)
at org.junit.internal.runners.JUnit4ClassRunner.run(JUnit4ClassRunner.java:42)
at org.junit.runner.JUnitCore.run(JUnitCore.java:130)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
[DEBUG 18-03-07 16:56:49 debug消息 debug hello hi] Caller+0 at com.elk.log.ESLogTest.testLog(ESLogTest.java:16)
[WARN 18-03-07 16:56:49 警告警告] Caller+0 at com.elk.log.ESLogTest.testLog(ESLogTest.java:17)
[DEBUG 18-03-07 16:56:50 Request and operation succeeded] Caller+0 at io.searchbox.action.AbstractAction.createNewElasticSearchResult(AbstractAction.java:75)
io.searchbox.core.DocumentResult@4421d75d
[DEBUG 18-03-07 16:56:51 Request and operation succeeded] Caller+0 at io.searchbox.action.AbstractAction.createNewElasticSearchResult(AbstractAction.java:75)
io.searchbox.core.DocumentResult@25a5426c
[DEBUG 18-03-07 16:56:51 Request and operation succeeded] Caller+0 at io.searchbox.action.AbstractAction.createNewElasticSearchResult(AbstractAction.java:75)
io.searchbox.core.DocumentResult@5fd33fbc
[DEBUG 18-03-07 16:56:51 Request and operation succeeded] Caller+0 at io.searchbox.action.AbstractAction.createNewElasticSearchResult(AbstractAction.java:75)
io.searchbox.core.DocumentResult@25a52369
可以说都是我们想要的信息。
我们再回到Kibana：
image.png
可以看到，测试上的4条日志信息已经可以插到ES上。
然后再看看错误的那条日志信息：
image.png
报错的类，方法，行数，报错内容。。。都一一列出来。
任务完成，到此，我们可以在不影响其他代码的情况下，将日志插入到ES里，在以后的编程中可以快速定位问题。
扩展：
我们看看除了DBAppender实现Appender这条线外，还有哪些类走这条线：
image.png
我们看到实现Appender接口的就有两个类：AppenderBase和UnsynchronizedAppenderBase。
从字面上的意思就是一个同步的，一个异步的，我们用对比工具对比一下，发现这两个类几乎没什么区别，最大不同就是AppenderBase的doAppend方法多了synchronized的锁。
像插到数据库日志的DBAppender，控制台打印日志的ConsoleAppender。。。就会走异步的UnsynchronizedAppenderBase这条线。
像SMTPAppender，JMSAppenderBase，SocketAppenderBase。。。就会走AppenderBase这条线。
所以像DBAppender和DBAppenderBase，这两个类可以合成一个整体，直接继承UnsynchronizedAppenderBase，走异步这条线。
-----------------------谢谢观看，希望本文对你有帮助----------------------------
网友评论