基于ELK+Kafka构建分布式日志采集系统

分类：Kafka作者：Y了个J

文章目录

一、分布式日志采集产生背景

1.传统日志采集存在哪些缺点
2.elk+kafka日志采集的原理
3.基于docker compose 安装elk+kafka环境
4.基于AOP+并发队列实现日志的采集

在传统项目中，如果在生产环境中，有多台不同的服务器集群，如果生产环境需要通过日志定位项目的Bug的话，需要在每台节点上使用传统的命令方式查询，这样效率非常低下。
因此我们需要集中化的管理日志，ELK则应运而生。

在这里插入图片描述


传统方式服务器搜索日志命令：tail -200f 日志文件名称

二、ELK+Kafka组成

Elk E= ElasticSeach（存储日志信息）
l Logstash(搬运工)
K Kibana连接到我们ElasticSeach图形化界面查询日志
Elk+kafka实现分布式日志采集

为什么需要将日志存储在ElasticSeach 而不是MySQL中呢
ElasticSeach 底层使用到倒排索引存储数据 ，在 搜索日志效率比mysql要高的。

elk+kafka原理

  1. springboot项目会基于aop的方式拦截系统中日志，请求与响应日志信息—前置或者环绕通知；
  2. 将该日志投递到我们kafka中 注意该过程一定要是异步的形式，如果是同步形式会影响到整体
    接口的响应速度。
  3. Logstash 数据源—kafka 订阅kafka的主题 获取日志消息内容
  4. Logstash 在将日志消息内容输出到es中存放
  5. 开发者使用Kibana连接到ElasticSeach 查询存储日志内容。

为什么ELK需要结合Kafka

如果只整合elk 不结合kafka这样的话 每个服务器节点上都会安装Logstash做读写日志IO操作，可能性能不是很好，而且比较冗余。

在这里插入图片描述

ELK+Kafka环境构建

整个环境采用 docker compose 来实现构建
注意：环境 cpu 多核 内存 4GB以上

创建docker-compose.yml

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
    restart: always
  kafka:
    image: wurstmeister/kafka:2.12-2.3.0
    ports:
      - "9092:9092"
    environment:
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.75.129:9092
      - KAFKA_LISTENERS=PLAINTEXT://:9092
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    restart: always
  kafka-manager:  
    image: sheepkiller/kafka-manager                ## 镜像：开源的web管理kafka集群的界面
    environment:
        ZK_HOSTS: 192.168.75.129                   ## 修改:宿主机IP
    ports:  
      - "9001:9000"                                 ## 暴露端口
  elasticsearch:
    image: daocloud.io/library/elasticsearch:6.5.4
    restart: always
    container_name: elasticsearch
    environment:
       - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ports:
    - 9200:9200
  kibana:
    image: daocloud.io/library/kibana:6.5.4
    restart: always
    container_name: kibana
    ports:
    - 5601:5601
    environment:
      - elasticsearch_url=http://192.168.75.129:9200
    depends_on:
      - elasticsearch

6.关闭防火墙
systemctl stop firewalld
service iptables stop
7.docker-compose up -d 执行即可。

验证elk+kafka 环境

访问:zk 192.168.75.143:2181
访问:es http://192.168.75.143:9200/
访问:kibana http://192.168.75.143:5601/app/kibana#/dev_tools/console

在这里插入图片描述


在这里插入图片描述

安装 logstash

上传logstash-6.4.3.tar.gz到服务中
tar zxvf logstash-6.4.3.tar.gz
cd logstash-6.4.3
bin/logstash-plugin install logstash-input-kafka
bin/logstash-plugin install logstash-output-elasticsearch

注意：安装
bin/logstash-plugin install logstash-input-kafka
bin/logstash-plugin install logstash-output-elasticsearch
本机电脑需要有JDK的环境，如果没有JDK环境直接安装 logstash-input-kafka 或者logstash-output-elasticsearch会报错的。

在这里插入图片描述


在这里插入图片描述

在 logstash config 目录 创建 kafka.conf

input {
  kafka {
    bootstrap_servers => "192.168.75.143:9092"
    topics => "mayikt-log"
 }
}
filter {
  #Only matched data are send to output.
}
output {
   elasticsearch {
    action => "index"          #The operation on ES
    hosts  => "192.168.75.143:9200"   #ElasticSearch host,can be array.
    index  => "my_logs"         #The index to write data to.
  }  
}

进入logstash bin 目录 执行 ./logstash -f …/config/kafka.conf

三、springboot项目整合elk+kafka

maven依赖


    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-databind</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.66</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-aop</artifactId>
        </dependency>
        <dependency>
            <groupId>commons-lang</groupId>
            <artifactId>commons-lang</artifactId>
            <version>2.6</version>
        </dependency>
    </dependencies>

aop拦截系统日志


import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import javax.servlet.http.HttpServletRequest;

import com.alibaba.fastjson.JSONObject;

import com.mayikt.container.LogContainer;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

@Aspect
@Component
public class AopLogAspect {
    @Value("${server.port}")
    private String serverPort;

    // 申明一个切点 里面是 execution表达式
    @Pointcut("execution(* com.mayikt.api.service.*.*(..))")
    private void serviceAspect() {
    }

    //
    @Autowired
    private LogContainer logContainer;

    //
    // 请求method前打印内容
    @Before(value = "serviceAspect()")
    public void methodBefore(JoinPoint joinPoint) {
        ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder
                .getRequestAttributes();
        HttpServletRequest request = requestAttributes.getRequest();
        JSONObject jsonObject = new JSONObject();
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 设置日期格式
        jsonObject.put("request_time", df.format(new Date()));
        jsonObject.put("request_url", request.getRequestURL().toString());
        jsonObject.put("request_method", request.getMethod());
        jsonObject.put("signature", joinPoint.getSignature());
        jsonObject.put("request_args", Arrays.toString(joinPoint.getArgs()));
        // IP地址信息
        jsonObject.put("ip_addres", getIpAddr(request) + ":" + serverPort);
        JSONObject requestJsonObject = new JSONObject();
        requestJsonObject.put("request", jsonObject);
        jsonObject.put("request_time", df.format(new Date()));
        jsonObject.put("log_type", "info");
        // 将日志信息投递到kafka中
        String log = requestJsonObject.toJSONString();
//      ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send("mayikt-log",ctx);
        logContainer.addLog(log);
    }
    //
//    // 在方法执行完结后打印返回内容
//    @AfterReturning(returning = "o", pointcut = "serviceAspect()")
//    public void methodAfterReturing(Object o) {
//        ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder
//                .getRequestAttributes();
//        HttpServletRequest request = requestAttributes.getRequest();
//        JSONObject respJSONObject = new JSONObject();
//        JSONObject jsonObject = new JSONObject();
//        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 设置日期格式
//        jsonObject.put("response_time", df.format(new Date()));
//        jsonObject.put("response_content", JSONObject.toJSONString(o));
//        // IP地址信息
//        jsonObject.put("ip_addres", getIpAddr(request) + ":" + serverPort);
//        jsonObject.put("log_type", "info");
//        respJSONObject.put("response", jsonObject);
//        // 将日志信息投递到kafka中
////      kafkaTemplate.send("mayikt-log",respJSONObject.toJSONString());
////        logContainer.put(respJSONObject.toJSONString());
//    }
//
//

    /**
     * 异常通知
     *
     * @param point
     */
    @AfterThrowing(pointcut = "serviceAspect()", throwing = "e")
    public void serviceAspect(JoinPoint point, Exception e) {
        ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder
                .getRequestAttributes();
        HttpServletRequest request = requestAttributes.getRequest();
        JSONObject jsonObject = new JSONObject();
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 设置日期格式
        jsonObject.put("request_time", point.getSignature());
        jsonObject.put("request_args", Arrays.toString(point.getArgs()));
        jsonObject.put("error", e.toString());
        // IP地址信息
        jsonObject.put("ip_addres", getIpAddr(request) + ":" + serverPort);
        jsonObject.put("log_type", "info");
        JSONObject requestJsonObject = new JSONObject();
        requestJsonObject.put("request", jsonObject);
        // 将日志信息投递到kafka中
        String log = requestJsonObject.toJSONString();
        logContainer.addLog(log);
    }

    //
    public static String getIpAddr(HttpServletRequest request) {
        //X-Forwarded-For（XFF）是用来识别通过HTTP代理或负载均衡方式连接到Web服务器的客户端最原始的IP地址的HTTP请求头字段。
        String ipAddress = request.getHeader("x-forwarded-for");
        if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
            ipAddress = request.getHeader("Proxy-Client-IP");
        }
        if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
            ipAddress = request.getHeader("WL-Proxy-Client-IP");
        }
        if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
            ipAddress = request.getRemoteAddr();
            if (ipAddress.equals("127.0.0.1") || ipAddress.equals("0:0:0:0:0:0:0:1")) {
                //根据网卡取本机配置的IP
                InetAddress inet = null;
                try {
                    inet = InetAddress.getLocalHost();
                } catch (UnknownHostException e) {
                    e.printStackTrace();
                }
                ipAddress = inet.getHostAddress();
            }
        }
        //对于通过多个代理的情况，第一个IP为客户端真实IP,多个IP按照','分割
        if (ipAddress != null && ipAddress.length() > 15) { //"***.***.***.***".length() = 15
            if (ipAddress.indexOf(",") > 0) {
                ipAddress = ipAddress.substring(0, ipAddress.indexOf(","));
            }
        }
        return ipAddress;
    }
}

配置文件内容

spring:
  application:
    ###服务的名称
    name: mayikt-elkkafka
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
  kafka:
    bootstrap-servers: 192.168.75.143:9092 #指定kafka server的地址，集群配多个，中间，逗号隔开
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: default_consumer_group #群组ID
      enable-auto-commit: true
      auto-commit-interval: 1000
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
server:
  port: 9000

在kibana页面查看日志

在这里插入图片描述

原文地址：https://blog.csdn.net/weixin_44146121/article/details/130424203

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。

相关推荐

Kafka的认证
文章浏览阅读4.1k次。kafka认证_kafka认证
Kafka之常用参数配置整理
文章浏览阅读4.8k次，点赞4次，收藏11次。kafka常用参数_kafka配置
Kafka生产者发送消息的流程
文章浏览阅读1.4k次，点赞25次，收藏10次。Kafka 生产者发送消息的流程涉及多个步骤，从消息的创建到成功存储在 Kafka 集群中。_kafka发送消息流程
第3、4章 Kafka 生产者 和 消费者 ——向 Kafka 写入数据 和读取数据
文章浏览阅读854次，点赞22次，收藏24次。点对点模型：适用于一对一的消息传递，具有高可靠性。发布/订阅模型：适用于广播消息给多个消费者，实现消息的广播。主题模型：适用于根据消息的主题进行灵活的过滤和匹配，处理复杂的消息路由需求。
kafka消息队列的初步探索
文章浏览阅读1.5k次，点赞2次，收藏3次。kafka 自动配置在KafkaAutoConfiguration
【可视化工具】Kafka-Offset Explorer使用
文章浏览阅读1.3w次，点赞6次，收藏33次。Offset Explorer（以前称为Kafka Tool）是一个用于管理和使Apache Kafka ®集群的GUI应用程序。它提供了一个直观的UI，允许人们快速查看Kafka集群中的对象以及存储在集群主题中的消息。它包含面向开发人员和管理员的功能。二、环境信息系统环境：windows 10版本：2.2Kafka版本：Kafka2.0.0三、安装和使用3.1 下载Offset Explorer 和安装下载到本地的 .exe文件Next安装路径 ，Next。_offset explorer
【云原生进阶之PaaS中间件】第三章Kafka-4.3.2-broker网络模型
文章浏览阅读1.3k次，点赞12次，收藏19次。kafka broker 在启动的时候，会根据你配置的listeners 初始化它的网络组件，用来接收外界的请求，这个listeners你可能没配置过，它默认的配置是listeners=PLAINTEXT://:9092就是告诉kafka使用哪个协议，监听哪个端口，如果我们没有特殊的要求的话，使用它默认的配置就可以了，顶多是修改下端口这块。
Kafka: 详解、使用教程和示例
文章浏览阅读1.3k次，点赞2次，收藏2次。Kafka 是一个强大的分布式流处理平台，用于实时数据传输和处理。通过本文详细的介绍、使用教程和示例，你可以了解 Kafka 的核心概念、安装、创建 Topic、使用生产者和消费者，从而为构建现代分布式应用打下坚实的基础。无论是构建实时数据流平台、日志收集系统还是事件驱动架构，Kafka 都是一个可靠、高效的解决方案。_博客系统怎么使用kafka
【Spring从成神到升仙系列 四】从源码分析 Spring 事务的来龙去脉
文章浏览阅读3.5k次，点赞42次，收藏56次。对于Java开发者而言，关于 Spring ，我们一般当做黑盒来进行使用，不需要去打开这个黑盒。但随着目前程序员行业的发展，我们有必要打开这个黑盒，去探索其中的奥妙。本期 Spring 源码解析系列文章，将带你领略 Spring 源码的奥秘。本期源码文章吸收了之前 Kafka 源码文章的错误，将不再一行一行的带大家分析源码，我们将一些不重要的分当做黑盒处理，以便我们更快、更有效的阅读源码。废话不多说，发车！
Kafka篇——Kafka消费者端常见配置，涵盖自动手动提交offset、poll消息细节、健康状态检查、新消费组消费offset规则以及指定分区等技术点配置，全面无死角，一篇文章拿下！
文章浏览阅读1.1k次，点赞14次，收藏16次。一、自动提交offset1、概念Kafka中默认是自动提交offset。消费者在poll到消息后默认情况下，会自动向Broker的_consumer_offsets主题提交当前主题-分区消费的偏移量2、自动提交offset和手动提交offset流程图3、在Java中实现配置4、自动提交offset问题自动提交会丢消息。因为如果消费者还没有消费完poll下来的消息就自动提交了偏移量，那么此时消费者挂了，于是下一个消费者会从已经提交的offset的下一个位置开始消费消息。_kafka中自动提交offsets
分布式 - 消息队列Kafka：Kafka生产者架构和配置参数
文章浏览阅读1.6k次。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，这个时候KafkaProducer的send（）方法调用要么被阻塞，要么抛出异常，这个取决于参数max.block.ms的配置，此参数的默认值为60000，即60秒。在默认情况下，生产者发送的消息是未经压缩的。如果应用程序调用send()方法的速度超过生产者将消息发送给服务器的速度，那么生产者的缓冲空间可能会被耗尽，后续的send()方法调用会等待内存空间被释放，如果在max.block.ms之后还没有可用空间，就抛出异常。_kafka producer 参数
Kafka快速入门(最新版3.6.0)
文章浏览阅读2.9k次，点赞3次，收藏10次。kafka解决通信问题_kafka3.6
nginx+rsyslog+kafka+clickhouse+grafana 实现nginx 网关监控
文章浏览阅读1.5k次，点赞9次，收藏11次。上面都配置完了之后可以先验证下，保证数据最终到ck，如果有问题，需要再每个节点调试，比如先调试nginx->rsyslog ，可以先不配置kafka 输出，配置为console或者文件输出都可以，具体这里就不写了。这里做了一个类型转换，因为nginx，request-time 单位是s，我想最终呈现在grafana 中是ms，所以这里做了转换，当然grafana中也可以做。kafka 相关部署这里不做赘述，只要创建一个topic 就可以。
Kafka中的enable-auto-commit和auto-commit-interval配置
文章浏览阅读1.4k次，点赞22次，收藏16次。Kafka中的enable-auto-commit和auto-commit-interval配置_auto-commit-interval
thingsboard规则链调用外部API——kafka
文章浏览阅读742次。thingsboard规则链调用外部 kafka_thingsboard kafka
Kafka 简介
文章浏览阅读1.3k次，点赞18次，收藏22次。Kafka_简介
从kafka如何保证数据一致性看通常数据一致性设计
文章浏览阅读1.1k次，点赞16次，收藏14次。在数据库系统中有个概念叫事务，事务的作用是为了保证数据的一致性，意思是要么数据成功，要么数据失败，不存在数据操作了一半的情况，这就是数据的一致性。在很多系统或者组件中，很多场景都需要保证数据的一致性，有的是高度的一致性。特别是在交易系统等这样场景。有些组件的数据不一定需要高度保证数据的一致性，比如日志系统。本节从从kafka如何保证数据一致性看通常数据一致性设计。
元数据管理-Atlas的介绍和使用(集成Hive、Solr、Kafka、Kerberos)
文章浏览阅读1.4k次。概述介绍架构发展架构原理类型系统介绍类型hive_table类型介绍DataSet类型定义Asset类型定义Referenceable类型定义Process类型定义Entities(实体)Attributes(属性)安装安装环境准备安装Solr-7.7.3安装Atlas2.1.0Atlas配置Atlas集成HbaseAtlas集成SolrAtlas集成KafkaAtlas Server配置Kerberos相关配置Atlas集成HiveAtlas启动Atlas使用Hive元数据初次导入Hive元数据增量同步。_atlas元数据管理
zookeeper + kafka
文章浏览阅读659次。Zookeeper是一个开源的分布式服务管理框架。存储业务服务节点元数据及状态信息，并负责通知再 ZooKeeper 上注册的服务几点状态给客户端。
Kafka-Kraft 模式架构部署
文章浏览阅读1.4k次。Kafka-Kraft 模式架构部署_kafka kraft部署
pythonJavaScriptjavaHTMLPHPreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-boot