【玩转Kafka】海量日志收集实战之架构介绍与日志输出

海量日志收集实战

话不多说,看架构图:
在这里插入图片描述
最左侧,是Beats,它主要是用于收集日志的,比如这个Filebeat它的底层是用erlang语言写的,性能非常好,其实我们的系统中打印出来的日志,都会用这个Filebeat给我们抓取出来。
在我们这个架构中,Filebeat的主要作用就是把我们的日志搜集出来并转储到Kafka。

然后看Kafka的右边,是Logstash,它主要是用日志做一个过滤,然后它会把你过滤后的数据发送到Elasticsearch里。

最后利用Kibana进行展示。

接下来,我们就来说一下我们要做的事情,先看下这张图:
在这里插入图片描述
我们这了采用的是Log4j2,为什么不采用SpringBoot默认集成的logback?
这是因为Log4j2性能会更好一些,因为它的底层是基于无锁并行框架Disruptor做的,关于Disruptor大家可以看看这篇文章《java高阶编程之无锁并行计算框架——Disruptor初识》,当然,它想要发挥出威力还是有一定前提的,就是要求你配置一定要高,因为Disruptor比较耗内存和CPU,所以你想要日志收集的更实时,那对应的你应用服务的性能和配置也要高一些,并且也要经过一些压测。

这里我们看一下图中两个粉色的app.log与error.log,这里我们把它分成两个对应的log日志文件,app.log在这里我们可以存储一个全量的log日志,也就是说无论我们的业务打什么样级别的日志,都会存储到app.log里。然后error.log,只要出现异常就会存储在error.log里。这样的的目的主要是用于我们后续做一些告警和分析,所以我们后续的一些告警会通过对这个error.log做一些手段,比如定时任务做定时抓取。

最后,我们把这些日志通过filebeat抓取出来,转储到Kafka中,然后Kafka收到消息后,他对应的消费者也就是Logstash,它得到消息后在发送给Elasticsearch。

以上就是这个架构的整体设计。

日志输出

我们直接开拔,打开Idea,每天coding十小时,健康工作一百年。

首先引入依赖:

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.5.RELEASE</version>
        <relativePath/>
    </parent>
<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-logging</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-log4j2</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.58</version>
        </dependency>
        <dependency>
            <groupId>com.lmax</groupId>
            <artifactId>disruptor</artifactId>
            <version>3.3.4</version>
        </dependency>
    </dependencies>

工具类:
这是Dubbo底层用到的一个工具类,可以拿来直接使用。

public class NetUtil {

    public static String normalizeAddress(String address) {
        String[] blocks = address.split("[:]");
        if (blocks.length > 2) {
            throw new IllegalArgumentException(address + " is invalid");
        }
        String host = blocks[0];
        int port = 80;
        if (blocks.length > 1) {
            port = Integer.valueOf(blocks[1]);
        } else {
            address += ":" + port; //use default 80
        }
        String serverAddr = String.format("%s:%d", host, port);
        return serverAddr;
    }

    public static String getLocalAddress(String address) {
        String[] blocks = address.split("[:]");
        if (blocks.length != 2) {
            throw new IllegalArgumentException(address + " is invalid address");
        }
        String host = blocks[0];
        int port = Integer.valueOf(blocks[1]);

        if ("0.0.0.0".equals(host)) {
            return String.format("%s:%d", NetUtil.getLocalIp(), port);
        }
        return address;
    }

    private static int matchedIndex(String ip, String[] prefix) {
        for (int i = 0; i < prefix.length; i++) {
            String p = prefix[i];
            if ("*".equals(p)) { //*, assumed to be IP
                if (ip.startsWith("127.") ||
                        ip.startsWith("10.") ||
                        ip.startsWith("172.") ||
                        ip.startsWith("192.")) {
                    continue;
                }
                return i;
            } else {
                if (ip.startsWith(p)) {
                    return i;
                }
            }
        }

        return -1;
    }

    public static String getLocalIp(String ipPreference) {
        if (ipPreference == null) {
            ipPreference = "*>10>172>192>127";
        }
        String[] prefix = ipPreference.split("[> ]+");
        try {
            Pattern pattern = Pattern.compile("[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+");
            Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
            String matchedIp = null;
            int matchedIdx = -1;
            while (interfaces.hasMoreElements()) {
                NetworkInterface ni = interfaces.nextElement();
                Enumeration<InetAddress> en = ni.getInetAddresses();
                while (en.hasMoreElements()) {
                    InetAddress addr = en.nextElement();
                    String ip = addr.getHostAddress();
                    Matcher matcher = pattern.matcher(ip);
                    if (matcher.matches()) {
                        int idx = matchedIndex(ip, prefix);
                        if (idx == -1) continue;
                        if (matchedIdx == -1) {
                            matchedIdx = idx;
                            matchedIp = ip;
                        } else {
                            if (matchedIdx > idx) {
                                matchedIdx = idx;
                                matchedIp = ip;
                            }
                        }
                    }
                }
            }
            if (matchedIp != null) return matchedIp;
            return "127.0.0.1";
        } catch (Exception e) {
            return "127.0.0.1";
        }
    }

    public static String getLocalIp() {
        return getLocalIp("*>10>172>192>127");
    }

    public static String remoteAddress(SocketChannel channel) {
        SocketAddress addr = channel.socket().getRemoteSocketAddress();
        String res = String.format("%s", addr);
        return res;
    }

    public static String localAddress(SocketChannel channel) {
        SocketAddress addr = channel.socket().getLocalSocketAddress();
        String res = String.format("%s", addr);
        return addr == null ? res : res.substring(1);
    }

    public static String getPid() {
        RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
        String name = runtime.getName();
        int index = name.indexOf("@");
        if (index != -1) {
            return name.substring(0, index);
        }
        return null;
    }

    public static String getLocalHostName() {
        try {
            return (InetAddress.getLocalHost()).getHostName();
        } catch (UnknownHostException uhe) {
            String host = uhe.getMessage();
            if (host != null) {
                int colon = host.indexOf(':');
                if (colon > 0) {
                    return host.substring(0, colon);
                }
            }
            return "UnknownHost";
        }
    }
}

这个是FastJson通用的工具类:

public class FastJsonConvertUtil {

    private static final SerializerFeature[] featuresWithNullValue = { SerializerFeature.WriteMapNullValue, SerializerFeature.WriteNullBooleanAsFalse,
            SerializerFeature.WriteNullListAsEmpty, SerializerFeature.WriteNullNumberAsZero, SerializerFeature.WriteNullStringAsEmpty };

    /**
     * <B>方法名称:</B>将JSON字符串转换为实体对象<BR>
     * <B>概要说明:</B>将JSON字符串转换为实体对象<BR>
     * @param data JSON字符串
     * @param clzss 转换对象
     * @return T
     */
    public static <T> T convertJSONToObject(String data, Class<T> clzss) {
        try {
            T t = JSON.parseObject(data, clzss);
            return t;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * <B>方法名称:</B>将JSONObject对象转换为实体对象<BR>
     * <B>概要说明:</B>将JSONObject对象转换为实体对象<BR>
     * @param data JSONObject对象
     * @param clzss 转换对象
     * @return T
     */
    public static <T> T convertJSONToObject(JSONObject data, Class<T> clzss) {
        try {
            T t = JSONObject.toJavaObject(data, clzss);
            return t;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * <B>方法名称:</B>将JSON字符串数组转为List集合对象<BR>
     * <B>概要说明:</B>将JSON字符串数组转为List集合对象<BR>
     * @param data JSON字符串数组
     * @param clzss 转换对象
     * @return List<T>集合对象
     */
    public static <T> List<T> convertJSONToArray(String data, Class<T> clzss) {
        try {
            List<T> t = JSON.parseArray(data, clzss);
            return t;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * <B>方法名称:</B>将List<JSONObject>转为List集合对象<BR>
     * <B>概要说明:</B>将List<JSONObject>转为List集合对象<BR>
     * @param data List<JSONObject>
     * @param clzss 转换对象
     * @return List<T>集合对象
     */
    public static <T> List<T> convertJSONToArray(List<JSONObject> data, Class<T> clzss) {
        try {
            List<T> t = new ArrayList<T>();
            for (JSONObject jsonObject : data) {
                t.add(convertJSONToObject(jsonObject, clzss));
            }
            return t;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * <B>方法名称:</B>将对象转为JSON字符串<BR>
     * <B>概要说明:</B>将对象转为JSON字符串<BR>
     * @param obj 任意对象
     * @return JSON字符串
     */
    public static String convertObjectToJSON(Object obj) {
        try {
            String text = JSON.toJSONString(obj);
            return text;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * <B>方法名称:</B>将对象转为JSONObject对象<BR>
     * <B>概要说明:</B>将对象转为JSONObject对象<BR>
     * @param obj 任意对象
     * @return JSONObject对象
     */
    public static JSONObject convertObjectToJSONObject(Object obj){
        try {
            JSONObject jsonObject = (JSONObject) JSONObject.toJSON(obj);
            return jsonObject;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * <B>方法名称:</B><BR>
     * <B>概要说明:</B><BR>
     * @param obj
     * @return
     */
    public static String convertObjectToJSONWithNullValue(Object obj) {
        try {
            String text = JSON.toJSONString(obj, featuresWithNullValue);
            return text;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
}

然后,我们再写一个工具类:

@Componet
public class InputMDC implements EnvironmentAware {

    private static Environment environment;
    @Override
    public void setEnvironment(Environment environment) {
        InputMDC.environment = environment;
    }

    public static void putMDC() {
        MDC.put("hostName",NetUtil.getLocalHostName());
        MDC.put("ip",NetUtil.getLocalIp());
        MDC.put("applicationName",environment.getProperty("spring.application.name"));
    }
}

创建Controller:

@Slf4j
@RestController
public class IndexController {

    @RequestMapping(value = "/index")
    public String index(){
        InputMDC.putMDC();
        log.info("我是一个info级别日志");
        log.warn("我是一个warn级别日志");
        log.error("我是一个error级别日志");
        return "idx";
    }
}

创建application.yml:

server:
  servlet:
    context-path: /
  port: 8001

spring:
  application:
    name: log-collector
  http:
    encoding:
      charset: UTF-8
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
    default-property-inclusion: NON_NULL

创建log4j2.xml

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="INFO" schema="Log4J-V2.0.xsd" monitorInterval="600" >
    <Properties>
        <Property name="LOG_HOME">logs</Property>
        <property name="FILE_NAME">collector</property>
        <property name="patternLayout">[%d{yyyy-MM-dd'T'HH:mm:ss.SSSZZ}] [%level{length=5}] [%thread-%tid] [%logger] [%X{hostName}] [%X{ip}] [%X{applicationName}] [%F,%L,%C,%M] [%m] ## '%ex'%n</property>
    </Properties>
    <Appenders>
        <Console name="CONSOLE" target="SYSTEM_OUT">
            <PatternLayout pattern="${patternLayout}"/>
        </Console>  
        <RollingRandomAccessFile name="appAppender" fileName="${LOG_HOME}/app-${FILE_NAME}.log" filePattern="${LOG_HOME}/app-${FILE_NAME}-%d{yyyy-MM-dd}-%i.log" >
          <PatternLayout pattern="${patternLayout}" />
          <Policies>
              <TimeBasedTriggeringPolicy interval="1"/>
              <SizeBasedTriggeringPolicy size="500MB"/>
          </Policies>
          <DefaultRolloverStrategy max="20"/>         
        </RollingRandomAccessFile>
        <RollingRandomAccessFile name="errorAppender" fileName="${LOG_HOME}/error-${FILE_NAME}.log" filePattern="${LOG_HOME}/error-${FILE_NAME}-%d{yyyy-MM-dd}-%i.log" >
          <PatternLayout pattern="${patternLayout}" />
          <Filters>
              <ThresholdFilter level="warn" onMatch="ACCEPT" onMismatch="DENY"/>
          </Filters>              
          <Policies>
              <TimeBasedTriggeringPolicy interval="1"/>
              <SizeBasedTriggeringPolicy size="500MB"/>
          </Policies>
          <DefaultRolloverStrategy max="20"/>         
        </RollingRandomAccessFile>            
    </Appenders>
    <Loggers>
        <!-- 业务相关 异步logger -->
        <AsyncLogger name="com.wcl.*" level="info" includeLocation="true">
          <AppenderRef ref="appAppender"/>
        </AsyncLogger>
        <AsyncLogger name="com.wcl.*" level="info" includeLocation="true">
          <AppenderRef ref="errorAppender"/>
        </AsyncLogger>       
        <Root level="info">
            <Appender-Ref ref="CONSOLE"/>
            <Appender-Ref ref="appAppender"/>
            <AppenderRef ref="errorAppender"/>
        </Root>         
    </Loggers>
</Configuration>

我们这里重点关注一下这个AsyncLogger,它这里就是底层使用了Disruptor,我们可以进到源码里看看:
在这里插入图片描述
点进去:
在这里插入图片描述
不难发现,它的log4j2的底层确实是采用了Disruptor,大家可以打个断点看看,这里就不详细探讨了。

项目结构:
在这里插入图片描述
我们启动一下项目,查看控制台:
在这里插入图片描述
可以看到,我们的控制台没有任何错误信息,说明已经正常启动起来了。

并且项目下多出来一个logs目录:
在这里插入图片描述

我们访问localhost:8001/index,在查看控制台

在这里插入图片描述
可以看到,我们想要的一些信息也通过InputMDC拿到了。

然后,我们人为制造一个异常:
在这里插入图片描述
访问后查看控制台:
在这里插入图片描述
并且我们也能在错误日志中看到这个错误信息:
在这里插入图片描述

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
THE END
分享
二维码
< <上一篇
下一篇>>