基于Flume+Log4j+Kafka的日志采集架构方案（下）

时间 2019-11-09

标签基于 flume+log4j+kafka flume log 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