基于Docker-Compose实现ELK+Kafka搭建分布式日志采集系统

ELK+Kafka搭建日志采集系统

  • ELK概述
  • 搭建与配置
    • docker-compose.yml
    • 配置日志采集规则
    • 启动服务
  • 模拟发送日志消息
    • 日志发送队列
    • 日志切面
    • 配置application.yaml
    • 发送日志消息
  • Kibana的使用
    • 创建索引模式
    • Discovery搜索数据
    • 可视化数据
  • ELK+RabbitMQ
    • 发送日志消息
    • 配置日志采集规则

ELK概述

ELK是指Elasticsearch、Logstash和Kibana这三个主要的开源软件工具，它们通常一起被用于实时日志分析和数据可视化。另外，在日志采集系统中，ELK通常结合Kafkka一起使用。

1、Elasticsearch

Elasticsearch是一个开源的分布式搜索和分析引擎，基于Lucene库构建。它设计用于处理大规模数据集，能够快速地进行全文搜索、结构化搜索、分析和实时数据处理。Elasticsearch具有高可扩展性和高可靠性，可以自动处理数据的分片和复制，支持分布式搜索和聚合操作。

2.Logstash：

Logstash是一个开源的数据收集和处理引擎，用于将各种类型的数据（如日志、事件、度量等）从多个来源采集、处理和传输到Elasticsearch或其他存储和分析工具。Logstash支持多种数据输入源和输出目的地，可以进行数据转换、标准化、过滤和增强，使数据具备一致性和结构性。

3.Kibana：

Kibana是一个开源的数据可视化平台，用于在Elasticsearch上创建和共享实时的数据可视化仪表盘。它提供了丰富的图表、表格、地图和仪表盘等可视化组件，使用户能够以直观的方式探索和分析数据。Kibana还支持交互式查询和过滤，能够快速演示和共享数据的见解。

4.Kafka

Kafka是数据缓冲队列。作为消息队列解耦了处理过程，同时提高了可扩展性。具有峰值处理能力，使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

搭建与配置

使用Docker Compose实现ELK（Elasticsearch、Logstash、Kibana）和Kafka日志采集

docker-compose.yml

创建docker-compose.yml文件，使用以下服务：

ZooKeeper：用于Kafka的依赖服务，监听在2181端口

Kafka：用于消息队列和日志采集，监听在9092端口，并连接到ZooKeeper

Elasticsearch：用于存储和索引日志数据，监听在9200端口

Logstash：用于从Kafka接收日志数据并转发到Elasticsearch

Kibana：用于可视化和检索日志数据，监听在5601端口，并连接到Elasticsearch

vim docker-compose.yml

version: '3.7'
services:
  zookeeper:
    image: zookeeper:3.8
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes  
    restart: always
  kafka:
    image: bitnami/kafka:3.3.2
    container_name: kafka1
    hostname: kafka
    volumes:
      - ./kafka_data:/bitnami/kafka  # 赋予kafka_data目前权限：chmod 777 kafka_data
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper  
    environment:
        KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181 #kafka链接的zookeeper地址
        KAFKA_ENABLE_KRAFT: no # 是否使用kraft，默认值：是,即Kafka替代Zookeeper
        KAFKA_CFG_LISTENERS: PLAINTEXT://:9092 # 定义kafka服务端socket监听端口,默认值：PLAINTEXT://:9092,CONTROLLER://:9093
        KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://192.168.30.30:9092 # 定义外网访问地址（宿主机ip地址和端口）,默认值：PLAINTEXT://:9092
        KAFKA_KRAFT_CLUSTER_ID: FDAF211E728140229F6FCDF4ADDC0B32 # 使用Kafka时的集群id，集群内的Kafka都要用这个id做初始化，生成一个UUID即可
        ALLOW_PLAINTEXT_LISTENER: yes # 允许使用PLAINTEXT监听器，默认false，不建议在生产环境使用
        KAFKA_HEAP_OPTS: -Xmx512M -Xms256M # 设置broker最大内存，和初始内存
        KAFKA_BROKER_ID: 1 # broker.id，必须唯一
    restart: always
  elasticsearch:
    image: elasticsearch:7.4.2
    container_name: elasticsearch
    hostname: elasticsearch
    volumes:
      - ./es_data:/usr/share/elasticsearch/data # 赋予es_data目前权限：chmod 777 es_data
    restart: always
    environment:
       - "discovery.type=single-node"
       - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ports:
    - "9200:9200"
  logstash:
    image: logstash:7.4.2
    container_name: logstash
    volumes:
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
      - ./logstash.yml:/usr/share/logstash/config/logstash.yml
    depends_on:
      - elasticsearch
    environment:
      LS_JAVA_OPTS: "-Xmx256m -Xms128m"
      ELASTICSEARCH_HOST: "http://192.168.30.30:9200"
  kibana:
    image: kibana:7.4.2
    restart: always
    container_name: kibana1
    ports:
    - 5601:5601
    environment:
      ELASTICSEARCH_URL: "http://192.168.30.30:9200"
    depends_on:
      - elasticsearch

配置日志采集规则

创建kafka的数据存储目录，并赋予目前权限：

chmod 777 kafka_data

创建ES的数据存储目录，并赋予目前权限：

chmod 777 es_data

创建logstash.yml配置文件，修改ES连接地址

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://192.168.30.30:9200" ]

创建logstash.conf配置文件，定义日志采集规则

input {
   kafka {
     bootstrap_servers => "192.168.30.30:9092"
     topics => "user_logs"
 }
}
filter {
  
}

output {
  elasticsearch {
    hosts  => "192.168.30.30:9200"
    index  => "user_logs"
 }
}

启动服务

在docker-compose.yml文件目录，运行以下命令来启动服务：

docker-compose up -d

所有容器都启动完毕后，可以通过访问http://192.168.30.30:5601/来访问Kibana，开始对日志数据进行可视化和查询分析
在这里插入图片描述

模拟发送日志消息

创建一个日志发送队列，配合线程，从队列中获取日志内容，然后以异步的形式发送消息到MQ

日志发送队列

@Component
public class LogDeque {
    /**
     * 本地队列
     */
    private static LinkedBlockingDeque<String> logMsgs = new LinkedBlockingDeque<>();
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void log(String msg) {
        logMsgs.offer(msg);
    }

    public LogDeque() {
        new LogThread().start();
    }

    /**
     * 创建线程,从队列中获取日志内容,然后以异步的形式发送消息到MQ
     */
    class LogThread extends Thread {
        @Override
        public void run() {
            while (true) {
                String msgLog = logMsgs.poll();
                if (!StringUtils.isEmpty(msgLog)) {
                    // 发送消息
                    kafkaTemplate.send("user_logs", msgLog);
                }
                // 避免cpu飙高的问题
                try {
                    Thread.sleep(200);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

日志切面

@Aspect
@Component
@Slf4j
public class AopLogAspect {
    @Autowired
    private LogDeque logDeque;

    /**
     * 申明一个切点 execution表达式
     */
    @Pointcut("execution(* cn.ybzy.demo.controller.*.*(..))")
    private void logAspect() {
    }

    /**
     * 请求method前打印内容
     *
     * @param joinPoint
     */
    @Before(value = "logAspect()")
    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_ip", request.getRemoteAddr());
        jsonObject.put("request_method", request.getMethod());
        jsonObject.put("request_args", Arrays.toString(joinPoint.getArgs()));

        // 将日志信息投递到MQ
        String logMsg = jsonObject.toJSONString();
        log.info("<AOP日志 ===》 MQ投递消息：{}>", logMsg);

        // 投递msg
        logDeque.log(logMsg);
    }
}

配置application.yaml

server:
  port: 8888

spring:
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://127.0.0.1:3306/demo?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowMultiQueries=true&serverTimezone=UTC
    username: root
    password: 123456



  application:
    # 服务的名称
    name: elkk
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
  kafka:
    bootstrap-servers: 192.168.30.30: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

发送日志消息

@Slf4j
@RestController
public class TestController {

    @RequestMapping("/test")
    public String test() {
        return "OK";
    }
}

Kibana的使用

要使用Kibana，需要通过配置一个或多个索引模式来告诉它你想探索的Elasticsearch索引，索引模式(index pattern)是kibana可视化的前提。它相当于告诉kibana要使用哪些索引作为数据进行可视化展示。

创建索引模式

一个索引模式标识一个或者多个你想要通过kiabna探索的Elasticsearch索引。Kibana会查找与指定模式匹配的索引名称。模式中的星号 (*) 匹配0个或者多个字符。

在左侧菜单找到management，然后点击index patterns --> create index pattern。通过输入index pattern的名称，kibana会自动显示匹配的索引，然后点击next

在这里插入图片描述

在配置设置Configure settings中， 选择索引中的时间维度的字段，这个时间字段是用来方便基于时间过滤数据用的。这里下拉菜单中选择@timestamp，单击创建索引模式

在这里插入图片描述

Discovery搜索数据

点击Discovery，可以查看到日志相关数据，也可以进行日志搜索
在这里插入图片描述

可视化数据

kibana自带了很多可视化的组件，方便对聚合后的结果进行可视化的展示。

左侧菜单选择Visualize，然后点击右边的+号
在这里插入图片描述

ELK+RabbitMQ

搭建ELK+Kafka日志采集系统后，同理可搭建ELK+RabbitMQ日志采集系统，他们实现方式类似，以下为参考：

发送日志消息

	@Autowired
	private RabbitTemplate rabbitTemplate;
	
	@Test
    public void logs(String logMsg)  {
        for (int i = 0; i < 500000; i++) {
            rabbitTemplate.convertAndSend("elk_logs_exchange", "user_logs",logMsg);
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

配置日志采集规则

创建logstash.conf配置文件，定义日志采集规则

input {
  rabbitmq {
    host => "192.168.30.30:5672"
    user => "work"
    password => "12345678"
    vhost => "/"
    queue => "user_logs"
    durable => true
    exchange => "elk_logs_exchange"
    key => "user.logs"
    codec => "json"
  }
}

output {
  elasticsearch {
    hosts => ["192.168.30.30:9200"]
    index => "base_logs"
  }
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/244932.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

Java IO 流详解

Java IO 流详解

Java IO 流详解 1 .Java IO概念 Java IO&#xff1a;即 Java 输入 / 输出系统。 Java 的 IO 模型设计非常优秀&#xff0c;它使用 Decorator (装饰者)模式&#xff0c;按功能划分 Stream &#xff0c;您可以动态装配 这些 Stream &#xff0c;以便获得您需要的功能。 Stream &…
阅读更多...
20 Redis进阶 - 运维监控

20 Redis进阶 - 运维监控

1、理解Redis监控 Redis运维和监控的意义不言而喻&#xff0c;可以以下三个方面入手 1.首先是Redis自身提供了哪些状态信息&#xff0c;以及有哪些常见的命令可以获取Redis的监控信息; 2.一些常见的UI工具可以可视化的监控Redis; 3.理解Redis的监控体系;2、Redis自身状态及命…
阅读更多...
【Java反射详解】

【Java反射详解】

Java反射详解 &#x1f38a;专栏【Java】 &#x1f354;喜欢的诗句&#xff1a;关山难越&#xff0c;谁悲失路之人。 萍水相逢&#xff0c;尽是他乡之客。 &#x1f386;音乐分享【Counting Stars 】 欢迎并且感谢大家指出问题&#x1f970; 1.什么是反射 所谓的反射就是java…
阅读更多...
365锦鲤助手 砍价小程序源码 流量主引流裂变

365锦鲤助手 砍价小程序源码 流量主引流裂变

源码介绍 修改版365锦鲤 助手&#xff0c; 砍价小程序源码 流量主引流裂变 拼多多商品快速丰富产品内容满足广大用户需求&#xff1b;流量矩阵让流量都进你的圈子飞起来&#xff1b;长期盈利、项目稳定 1.后台安装微擎 2安装应用 后台打包上传
阅读更多...
PPINN Parareal physics-informed neural network for time-dependent PDEs

PPINN Parareal physics-informed neural network for time-dependent PDEs

论文阅读&#xff1a;PPINN Parareal physics-informed neural network for time-dependent PDEs PPINN Parareal physics-informed neural network for time-dependent PDEs简介方法PPINN加速分析 实验确定性常微分方程随机常微分方程Burgers 方程扩散反应方程 总结 PPINN Par…
阅读更多...
Npm安装vue3报错(node:25436) MaxListenersExceededWarning:

Npm安装vue3报错(node:25436) MaxListenersExceededWarning:

运行命令安装vue3时 npm create vuelatest 报了错误(node:25436) MaxListenersExceededWarning: (忘记截报错的图了&#xff0c;后面还有一大串英文) 搞了很久发现是网络的原因&#xff0c;我没有修改镜像地址&#xff0c;导致访问很慢&#xff0c;于是去npmmirror 镜像站 …
阅读更多...
字符雨canvas

字符雨canvas

整体思路&#xff1a; 确定好字符雨的具体字符是什么&#xff0c;需要多少行多少列这里是写死的其实也可以用循环加随机的方式生成不一样的字符雨&#xff0c;行列也可以读一下宽度然后做一下出发算一下也行首先得有一张画布搞起&#xff0c;然后循环列数去绘画字符定时器循环…
阅读更多...
泰森多边形半平面求交 - 洛谷 - P3297 [SDOI2013] 逃考

泰森多边形半平面求交 - 洛谷 - P3297 [SDOI2013] 逃考

欢迎关注更多精彩 关注我&#xff0c;学习常用算法与数据结构&#xff0c;一题多解&#xff0c;降维打击。 往期相关背景半平面求交 点击前往 voronoi 图求解点击前往 题目大意 题目链接 https://www.luogu.com.cn/problem/P3297 小杨家是一个矩阵&#xff0c;所有亲戚都在…
阅读更多...
Monkey工具之fastbot-iOS实践

Monkey工具之fastbot-iOS实践

背景 目前移动端App上线后 crash 率比较高&#xff0c; 尤其在iOS端。我们需要一款Monkey工具测试App的稳定性&#xff0c;更早的发现crash问题并修复。 去年移动开发者大会上有参加 fastbot 的分享&#xff0c;所以很自然的就想到Fastbot工具。 Fastbot-iOS安装配置 准备工…
阅读更多...
变电站蓄电池在线监测系统（论文+源码）

变电站蓄电池在线监测系统（论文+源码）

1. 系统设计 本次课题为变电站蓄电池在线监测系统的设计&#xff0c;其系统架构如图3.1所示&#xff0c;包括了主控制器STC89C52单片机&#xff0c;液晶显示器LCD1602,模数转换器ADC0832&#xff0c;电流传感器ACS712&#xff0c;分压电阻&#xff0c;蜂鸣器以及温度传感器。在…
阅读更多...
实用篇 | 3D建模中Blender软件的下载及使用[图文详情]

实用篇 | 3D建模中Blender软件的下载及使用[图文详情]

本文基于数字人系列的3D建模工具Blender软件的安装及使用&#xff0c;还介绍了图片生成3D模型的AI工具~ 目录 1.Blender的下载 2.Blender的使用 3.安装插件(通过压缩包安装) 4.实例 4.1.Blender使用MB-Lab插件快速人体模型建构 4.1.1.点击官网&#xff0c;进行下载 4.1.…
阅读更多...
消息可靠性保证

消息可靠性保证

回顾RabbitMQ的消息传递过程 如图所示&#xff0c;发生消息丢失的可能阶段也就是生产者发送消息&#xff0c;时rabbitmq存储消息时&#xff0c;消费者消费消息时。项目源码&#xff1a;gitee 生产者发送消息阶段 生产者发送消息时把交换机名写错生产者发送消息时把routingK…
阅读更多...
推荐一款好用的包含表格识别的OCR网站

推荐一款好用的包含表格识别的OCR网站

在当今数字化的时代&#xff0c;文字和表格识别已经成为了许多行业的关键技术。无论是处理大量的纸质文档&#xff0c;还是从网络上收集数据&#xff0c;OCR&#xff08;光学字符识别&#xff09;技术都扮演着重要的角色。然而&#xff0c;对于许多用户来说&#xff0c;OCR软件…
阅读更多...
Infobright列存数据库原理介绍

Infobright列存数据库原理介绍

简介 Infobright 是一个面向 OLAP 场景的开源列存数据库。比较容易找到代码的版本是 Infobright Community Edition 4.0.7&#xff0c;大概是 2006 年前后的代码。2016 年6 月&#xff0c;Infobright 决定停止开源1。由于它同时提供企业版和社区版&#xff0c;开源版本的功能相…
阅读更多...
斑马zebra目标检测数据集VOC+YOLO格式2300张

斑马zebra目标检测数据集VOC+YOLO格式2300张

斑马是由四百万年前的原马进化出来的&#xff0c;最早出现的斑马可能是细纹斑马。有关史前马科动物的化石现存于美国爱达荷州克文的克文化石床国家博物馆。斑马的史前马为“克文马”&#xff08;美洲斑马或者克文斑马&#xff09;&#xff0c;学名为“Equussimplicidens”&…
阅读更多...
WordPress VIP收费下载插件Erphpdown v17.0.1 开心版

WordPress VIP收费下载插件Erphpdown v17.0.1 开心版

会员推广下载专业版 WordPress插件&#xff08;erphpdown&#xff09;是模板兔开发的一款针对虚拟资源收费下载/付费下载/付费视频/收费查看/付费阅读/付费查看/VIP下载查看的插件&#xff0c;经过完美测试运行于wordpress 3.x-5.x版本。后续模板兔会增加更多实用的功能。 模板…
阅读更多...
docker部署go gin框架 Linux环境

docker部署go gin框架 Linux环境

目录 文章目的是什么 环境介绍 Linux 环境下 docker 部署 go gin 详细步骤 部署 gin 文章目的是什么 假设我们学习了 go 语言&#xff0c;在 Linux 上安装了 go 相关的程序&#xff0c;也能直接运行&#xff0c;使用以下命令&#xff1a; go run main.go 假如代码是这样的…
阅读更多...
跟着我学Python基础篇：08.集合和字典

跟着我学Python基础篇：08.集合和字典

往期文章 跟着我学Python基础篇&#xff1a;01.初露端倪 跟着我学Python基础篇&#xff1a;02.数字与字符串编程 跟着我学Python基础篇&#xff1a;03.选择结构 跟着我学Python基础篇&#xff1a;04.循环 跟着我学Python基础篇&#xff1a;05.函数 跟着我学Python基础篇&#…
阅读更多...
java设计模式学习之【代理模式】

java设计模式学习之【代理模式】

文章目录 引言代理模式简介定义与用途实现方式 使用场景优势与劣势在Spring框架中的应用图片加载示例代码地址 引言 在现实生活中&#xff0c;我们经常使用代理来处理我们不想直接参与或无法直接参与的事务&#xff0c;例如&#xff0c;使用律师来代表法庭上的案件。在软件开发…
阅读更多...
计算机网络——网络层——OSPF协议的介绍

计算机网络——网络层——OSPF协议的介绍

什么是 OSPF &#xff1f; OSPF 是一个基于链路状态的自治系统内部路由协议&#xff0c;在 TCP/IP 的网络层中进行路由选择&#xff0c;常用于构建大型企业网络或者服务上的骨干网络。在互联网核心路由器之间也可以使用。 OSPF 概述 OSPF 使用的是 Dijkstra&#xff08;最短…
阅读更多...
最新文章

Copyright @ 2022~2023