【306期】微服务 Spring Cloud 架构实现分布式日志采集方案

Java精选 2022-08-09

收录于合集

#SpringCloud 23 个

#面试题系列专栏 349 个

#Elasticsearch 16 个

点击上方“Java精选”，选择“设为星标”

别问别人为什么，多问自己凭什么！

下方有惊喜，留言必回，有问必答！

每一天进步一点点，是成功的开始...

一、简介

分布式应用必须有一套日志采集功能，目的是将分布在各个服务器节点上的应用日志文件采集到统一的服务器上，方便日志的查看。springCloud本身提供了基于elk的日志采集，但是由于使用logstash，会加大运维成本。这里将使用轻量级的方案。

二、思路

我们的目的是提供轻量级的日志采集来代替logstash，日志最终还是会存进Elasticsearch。为了能轻量级的实现日志采集，并且避免对代码的侵入，我们可以扩展Logback的appender，也可以扩展log4j的appender。这样我们使用slf4j来记录日志的时候，日志自动会保存到Elasticsearch中，并且不用修改任何业务代码。

三、自定义Logback appender

我们先来看一下Logback的appender的Uml图，我们可以发现两个对我们有借鉴意义的类

UnsynchronizedAppenderBase提供了异步的日志记录
DBAppender基于数据库的日志记录

这两个类还是比较简单的，具体的代码我就不详细解说了，请自行查阅

属性注入

基本实现逻辑从UnsynchronizedAppenderBase和DBAppender已经能够知道了，现在把我们需要的信息注入到Appender中，这里需要如下的知识

Logback标签注入属性

我们可以直接在Xml中用标签配置属性，这些标签只要名称和appender中的成员变量名一致，则会自动把标签中的属性注入到成员变量中。

我们举一个例子：

xml这样配置

<appender name="ES" class="com.luminroy.component.logger.appender.ElasticsearchAppender">
  <profile>test</profile>
  <esType>demo</esType>
  <withJansi>true</withJansi>
  <encoder>
   <pattern>${CONSOLE_LOG_PATTERN_IDE}</pattern>
   <charset>utf8</charset>
  </encoder>
 </appender>

其中ElasticsearchAppender是我们自己实现的Appender。这里有一个profile标签，我们需要ElasticsearchAppender中成员变量的名称和该标签名一致，这样就可以把test值注入到成员变量profile中。

protected String profile = ""; // 运行环境

Spring配置信息注入属性

有些信息可能已经在spring中做了配置，我们不想要重复的配置，这个时候我们可以用springProperty标签来进行设置。

scope：作用范围
name：名称
source：spring配置
defaultValue：默认值，必须要指定

然后在标签中用上面的name属性作为占位符，类中的成员变量名和标签名一致。

我们举一个例子：

xml这样配置

<springProperty scope="context" name="applicationName" source="spring.application.name"
     defaultValue=""/>
 <springProperty scope="context" name="profile" source="spring.profiles.active"
     defaultValue="default"/>
 
 <springProperty scope="context" name="esUserName" source="luminary.elasticsearch.username"
     defaultValue="elastic"/>
 
 <springProperty scope="context" name="esPassword" source="luminary.elasticsearch.password"
     defaultValue="123456"/>
 
 <springProperty scope="context" name="esServer" source="luminary.elasticsearch.server"
     defaultValue="127.0.0.1:9200"/>
 
 <springProperty scope="context" name="esMultiThreaded" source="luminary.elasticsearch.multiThreaded"
     defaultValue="true"/>
 
 <springProperty scope="context" name="esMaxTotalConnection" source="luminary.elasticsearch.maxTotalConnection"
     defaultValue="20"/>
 
 <springProperty scope="context" name="esMaxTotalConnectionPerRoute" source="luminary.elasticsearch.maxTotalConnectionPerRoute"
     defaultValue="5"/>
 
 <springProperty scope="context" name="esDiscoveryEnabled" source="luminary.elasticsearch.discoveryEnabled"
     defaultValue="true"/>
 
 <springProperty scope="context" name="esDiscorveryFrequency" source="luminary.elasticsearch.discorveryFrequency"
     defaultValue="60"/>

<appender name="ES" class="com.luminary.component.logger.appender.SpringElasticsearchAppender">
  <applicationName>${applicationName}</applicationName>
  <profile>${profile}</profile>
  <esType>demo</esType>
  <username>${esUserName}</username>
  <password>${esPassword}</password>
  <server>${esServer}</server>
  <multiThreaded>${esMultiThreaded}</multiThreaded>
  <maxTotalConnection>${esMaxTotalConnection}</maxTotalConnection>
  <maxTotalConnectionPerRoute>${esMaxTotalConnectionPerRoute}</maxTotalConnectionPerRoute>
  <discoveryEnabled>${esDiscoveryEnabled}</discoveryEnabled>
  <discorveryFrequency>${esDiscorveryFrequency}</discorveryFrequency>
 </appender>

yml这样配置

spring:
  application:
    name: logger-demo-server
 
luminary: 
  elasticsearch:
    username: elastic
    password: 123456
    server: 
      - 127.0.0.1:9200
    multiThreaded: true
    maxTotalConnection: 20
    maxTotalConnectionPerRoute: 5
    discoveryEnabled: true
    discorveryFrequency: 60

成员变量

@Setter
protected String esIndex = "java-log-#date#"; // 索引
@Setter
protected String esType = "java-log"; // 类型
@Setter
protected boolean isLocationInfo = true; // 是否打印行号
@Setter
protected String applicationName = "";
@Setter
protected String profile = ""; // 运行环境
@Setter
protected String esAddress = ""; // 地址

Logback代码注入属性

这里还有一种情况，有些属性需要在运行时才知道，或者运行时会改变。这就需要能动态注入属性。我们可以使用log4j的MDC类来解决。

我们可以通过相应的put，remove方法来动态设置属性。

比如：

MDC.put(TraceInfo.TRACE_ID_KEY, traceInfo.getTraceId());
MDC.put(TraceInfo.RPC_ID_KEY, traceInfo.getRpcId());

MDC.remove(TraceInfo.TRACE_ID_KEY);
MDC.remove(TraceInfo.RPC_ID_KEY);

获取属性值可以通过LoggingEvent的getMDCPropertyMap方法先获取属性的map，再根据键名从map中取出来。

比如：

private String getRpcId(LoggingEvent event) {
    Map<String, String> mdcPropertyMap = event.getMDCPropertyMap();
 return mdcPropertyMap.get("rpcId");
}
 
private String getTraceId(LoggingEvent event) {
 Map<String, String> mdcPropertyMap = event.getMDCPropertyMap();
 return mdcPropertyMap.get("traceId");
}

值得说明的是，mdcAdapter是一个静态的成员变量，但是它自身是线程安全的，我们可以看一下logback的实现

private Map<String, String> duplicateAndInsertNewMap(Map<String, String> oldMap) {
        Map<String, String> newMap = Collections.synchronizedMap(new HashMap<String, String>());
        if (oldMap != null) {
            // we don't want the parent thread modifying oldMap while we are
            // iterating over it
            synchronized (oldMap) {
                newMap.putAll(oldMap);
            }
        }
 
        copyOnThreadLocal.set(newMap);
        return newMap;
    }

Elasticsearch模板设计

最后日志保存在Elasticsearch中，我们希望索引名为java-log-${date}的形式，type名为实际的微服务名

最后我们对日志索引设置一个模板

举一个例子：

PUT _template/java-log
{
  "template": "java-log-*",
  "order": 0,
  "setting": {
    "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_field": {
            "match_mapping_type": "string",
            "path_match": "throwable",
            "mapping": {
              "norms": false,
              "type": "text",
              "analyzer": "ik_max_word",
              "search_analyzer": "ik_max_word"
            }
          }
        },
        {
          "string_field": {
            "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": {
       "applicationName": {
          "norms": false,
          "type": "text",
          "analyzer": "ik_max_word",
          "search_analyzer": "ik_max_word",
          "fields": {
           "keyword": {
             "type": "keyword",
              "ignore_above": 256
           }
          }
        },
        "profile": {
          "type": "keyword"
        },
        "host": {
          "type": "keyword"
        },
        "ip": {
          "type": "ip"
        },
        "level": {
          "type": "keyword"
        },
        "location": {
          "properties": {
            "line": {
               "type": "integer"
            }
          }
        },
        "dateTime": {
          "type": "date"
        },
        "traceId": {
          "type": "keyword"
        },
        "rpcId": {
          "type": "keyword"
        }
      }
    }
  }
}

示例代码

@Slf4j
public class ElasticsearchAppender<E> extends UnsynchronizedAppenderBase<E> implements LuminaryLoggerAppender<E> {
 
 private static final FastDateFormat SIMPLE_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd");
 
 private static final FastDateFormat ISO_DATETIME_TIME_ZONE_FORMAT_WITH_MILLIS = FastDateFormat.getInstance("yyyy-MM-dd'T'HH:mm:ss.SSSZZ");
 
 protected JestClient jestClient;
 
 private static final String CONFIG_PROPERTIES_NAME = "es.properties";
 
 // 可在xml中配置的属性
 @Setter
 protected String esIndex = "java-log-#date#"; // 索引
 @Setter
 protected String esType = "java-log"; // 类型
 @Setter
 protected boolean isLocationInfo = true; // 是否打印行号
 @Setter
 protected String applicationName = "";
 @Setter
 protected String profile = ""; // 运行环境
 @Setter
 protected String esAddress = ""; // 地址
 
 @Override
 public void start() {
  super.start();
  init();
 }
 
 @Override
 public void stop() {
  super.stop();
  // 关闭es客户端
  try {
   jestClient.close();
  } catch (IOException e) {
   addStatus(new ErrorStatus("close jestClient fail", this, e));
  }
 }
 
    @Override
    protected void append(E event) {
      if (!isStarted()) {
             return;
         }
 
      subAppend(event);
    }
 
    private void subAppend(E event) {
     if (!isStarted()) {
            return;
        }
     
     try {
            // this step avoids LBCLASSIC-139
            if (event instanceof DeferredProcessingAware) {
                ((DeferredProcessingAware) event).prepareForDeferredProcessing();
            }
            // the synchronization prevents the OutputStream from being closed while we
            // are writing. It also prevents multiple threads from entering the same
            // converter. Converters assume that they are in a synchronized block.
            save(event);
        } catch (Exception ioe) {
            // as soon as an exception occurs, move to non-started state
            // and add a single ErrorStatus to the SM.
            this.started = false;
            addStatus(new ErrorStatus("IO failure in appender", this, ioe));
        }
    }
    
    private void save(E event) {
     if(event instanceof LoggingEvent) {
      // 获得日志数据
   EsLogVO esLogVO = createData((LoggingEvent) event);
   // 保存到es中
   save(esLogVO);
     } else {
      addWarn("the error type of event!");
     }
    }
 
 private void save(EsLogVO esLogVO) {
  Gson gson = new Gson();
  String jsonString = gson.toString();
 
  String esIndexFormat = esIndex.replace("#date#", SIMPLE_FORMAT.format(Calendar.getInstance().getTime()));
  Index index = new Index.Builder(esLogVO).index(esIndexFormat).type(esType).build();
 
  try {
   DocumentResult result = jestClient.execute(index);
   addStatus(new InfoStatus("es logger result:"+result.getJsonString(), this));
  } catch (Exception e) {
   addStatus(new ErrorStatus("jestClient exec fail", this, e));
  }
 }
 
 private EsLogVO createData(LoggingEvent event) {
  EsLogVO esLogVO = new EsLogVO();
 
  // 获得applicationName
  esLogVO.setApplicationName(applicationName);
  
  // 获得profile
  esLogVO.setProfile(profile);
  
  // 获得ip
  esLogVO.setIp(HostUtil.getIP());
 
  // 获得hostName
  esLogVO.setHost(HostUtil.getHostName());
 
  // 获得时间
  long dateTime = getDateTime(event);
  esLogVO.setDateTime(ISO_DATETIME_TIME_ZONE_FORMAT_WITH_MILLIS.format(Calendar.getInstance().getTime()));
 
  // 获得线程
  String threadName = getThead(event);
  esLogVO.setThread(threadName);
 
  // 获得日志等级
  String level = getLevel(event);
  esLogVO.setLevel(level);
 
  // 获得调用信息
  EsLogVO.Location location = getLocation(event);
  esLogVO.setLocation(location);
 
  // 获得日志信息
  String message = getMessage(event);
  esLogVO.setMessage(message);
 
  // 获得异常信息
  String throwable = getThrowable(event);
  esLogVO.setThrowable(throwable);
 
  // 获得traceId
  String traceId = getTraceId(event);
  esLogVO.setTraceId(traceId);
 
  // 获得rpcId
  String rpcId = getRpcId(event);
  esLogVO.setRpcId(rpcId);
 
  return esLogVO;
 }
 
 private String getRpcId(LoggingEvent event) {
  Map<String, String> mdcPropertyMap = event.getMDCPropertyMap();
  return mdcPropertyMap.get("rpcId");
 }
 
 private String getTraceId(LoggingEvent event) {
  Map<String, String> mdcPropertyMap = event.getMDCPropertyMap();
  return mdcPropertyMap.get("traceId");
 }
 
 private String getThrowable(LoggingEvent event) {
  String exceptionStack = "";
  IThrowableProxy tp = event.getThrowableProxy();
  if (tp == null)
   return "";
 
  StringBuilder sb = new StringBuilder(2048);
  while (tp != null) {
 
   StackTraceElementProxy[] stackArray = tp.getStackTraceElementProxyArray();
 
   ThrowableProxyUtil.subjoinFirstLine(sb, tp);
 
   int commonFrames = tp.getCommonFrames();
   StackTraceElementProxy[] stepArray = tp.getStackTraceElementProxyArray();
   for (int i = 0; i < stepArray.length - commonFrames; i++) {
    sb.append("\n");
    sb.append(CoreConstants.TAB);
    ThrowableProxyUtil.subjoinSTEP(sb, stepArray[i]);
   }
 
   if (commonFrames > 0) {
    sb.append("\n");
    sb.append(CoreConstants.TAB).append("... ").append(commonFrames).append(" common frames omitted");
   }
 
   sb.append("\n");
 
   tp = tp.getCause();
  }
  return sb.toString();
 }
 
 private String getMessage(LoggingEvent event) {
  return event.getFormattedMessage();
 }
 
 private EsLogVO.Location getLocation(LoggingEvent event) {
  EsLogVO.Location location = new EsLogVO.Location();
  if(isLocationInfo) {
   StackTraceElement[] cda = event.getCallerData();
   if (cda != null && cda.length > 0) {
    StackTraceElement immediateCallerData = cda[0];
    location.setClassName(immediateCallerData.getClassName());
    location.setMethod(immediateCallerData.getMethodName());
    location.setFile(immediateCallerData.getFileName());
    location.setLine(String.valueOf(immediateCallerData.getLineNumber()));
   }
  }
  return location;
 }
 
 private String getLevel(LoggingEvent event) {
  return event.getLevel().toString();
 }
 
 private String getThead(LoggingEvent event) {
  return event.getThreadName();
 }
 
 private long getDateTime(LoggingEvent event) {
  return ((LoggingEvent) event).getTimeStamp();
 }
 
    private void init() {
  try {
   ClassLoader esClassLoader = ElasticsearchAppender.class.getClassLoader();
   Set<URL> esConfigPathSet = new LinkedHashSet<URL>();
   Enumeration<URL> paths;
   if (esClassLoader == null) {
    paths = ClassLoader.getSystemResources(CONFIG_PROPERTIES_NAME);
   } else {
    paths = esClassLoader.getResources(CONFIG_PROPERTIES_NAME);
   }
   while (paths.hasMoreElements()) {
    URL path = paths.nextElement();
    esConfigPathSet.add(path);
   }
 
   if(esConfigPathSet.size() == 0) {
    subInit();
    if(jestClient == null) {
     addWarn("没有获取到配置信息！");
     // 用默认信息初始化es客户端
     jestClient = new JestClientMgr().getJestClient();
    }
   } else {
 
    if (esConfigPathSet.size() > 1) {
     addWarn("获取到多个配置信息,将以第一个为准！");
    }
 
    URL path = esConfigPathSet.iterator().next();
    try {
     Properties config = new Properties();
     @Cleanup InputStream input = new FileInputStream(path.getPath());
     config.load(input);
     // 通过properties初始化es客户端
     jestClient = new JestClientMgr(config).getJestClient();
    } catch (Exception e) {
     addStatus(new ErrorStatus("config fail", this, e));
    }
 
   }
  } catch (Exception e) {
   addStatus(new ErrorStatus("config fail", this, e));
  }
 }
 
 @Override
 public void subInit() {
  // template method
 }
    
}