基于SpringBoot+logstash+redis实现日志异步集中存储

官网地址

简介

logstash是一个数据分析软件，主要目的是分析log日志。整一套软件可以当作一个MVC模型，logstash是controller层，Elasticsearch是一个model层，kibana是view层。首先将数据传给logstash，它将数据进行过滤和格式化（转成JSON格式），然后传给Elasticsearch进行存储、建搜索的索引，kibana提供前端的页面再进行搜索和图表可视化，它是调用Elasticsearch的接口返回的数据进行可视化。logstash和Elasticsearch是用Java写的，kibana使用node.js框架。

架构

应用场景

Logstash最常用于ELK（Elastic Stack+ logstash + kibana）中作为日志收集器使用

前提 已经安装好了redis并配置端口密码

下载并解压

1
mkdir -p /opt/logstash
2
cd /opt/logstash
3
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.7.0.tar.gz
4
tar -xvf logstash-7.7.0.tar.gz

配置

1
cd /opt/logstash/logstash-7.7.0/config
2
cp logstash-sample.conf my-logstash.conf
3
vim my-logstash.conf

• my-logstash.conf

  logstash最佳实践

  1
  input {
  2
          redis {
  3
                  data_type => "list"
  4
                  key => "logstash-list"
  5
                  host => "172.16.10.68"
  6
                  port => "6379"
  7
                  password => "12346"
  8
                  batch_count => 100
  9
                  threads => 1
  10
          }
  11
  }
  12
  
  13
  filter {
  14
       ruby {
  15
         code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
  16
       }
  17
       ruby {
  18
         code => "event.set('@timestamp',event.get('timestamp'))"
  19
       }
  20
      if [message] == "Completed 200 OK" {
  21
          drop {}
  22
      }
  23
  }
  24
  
  25
  output {
  26
      file{
  27
          path => "/opt/elk/logstash-7.1.1/logs/%{+yyyy}-%{+MM}-%{+dd}.log"
  28
          codec => line { format => "%{+yyyy}-%{+MM}-%{+dd} %{+HH}:%{+mm}:%{+ss} [%{thread}:%{requestId}] - [%{level}] %{message}"}
  29
      }
  30
        #这句话的意思是窗口输出内容 项目中要注释掉
  31
          stdout {codec => rubydebug}
  32
  }

  展开收起

• logback.xml

  1
  <?xml version="1.0" encoding="UTF-8"?>
  2
  <configuration>
  3
      <appender name="LOGSTASH" class="com.cwbase.logback.RedisAppender">
  4
          <host>172.16.10.68</host>
  5
          <port>6379</port>
  6
          <password>123456</password>
  7
          <key>logstash-list</key>
  8
          <callerStackIndex>0</callerStackIndex>
  9
          <additionalField>
  10
              <key>requestId</key>
  11
              <value>@{requestId}</value>
  12
          </additionalField>
  13
          <additionalField>
  14
              <key>requestTime</key>
  15
              <value>@{requestTime}</value>
  16
          </additionalField>
  17
      </appender>
  18
  
  19
      <appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">
  20
          <!-- 不丢失日志.默认的,如果队列的80%已满,则会丢弃TRACT、DEBUG、INFO级别的日志 -->
  21
          <discardingThreshold>0</discardingThreshold>
  22
          <!-- 更改默认的队列的深度,该值会影响性能.默认值为256 -->
  23
          <queueSize>5120</queueSize>
  24
          <appender-ref ref="LOGSTASH"/>
  25
      </appender>
  26
  
  27
      <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
  28
          <Target>System.out</Target>
  29
          <encoder>
  30
              <pattern>%d{yyyy-MM-dd HH:mm:ss} [ %t:%r ] [%X{requestId}] - [ %p ] %m%n</pattern>
  31
          </encoder>
  32
      </appender>
  33
      <root level="INFO">
  34
          <appender-ref ref="ASYNC"/>
  35
      </root>
  36
  </configuration>

  展开收起

启动

• 正常启动测试

  1
  /opt/logstash/logstash-7.7.0/bin/logstash -f /opt/logstash/logstash-7.7.0/config/my-logstash.conf

• 启动项目 并调用接口

• 效果图

  

  

• nohup 后台启动

  • 编写启动脚本文件 写入下面的内容

    1
    vim /opt/logstash/logstash-7.7.0/runLogstash.sh

    1
    nohup /opt/logstash/logstash-7.7.0/bin/logstash -f /opt/logstash/logstash-7.7.0/config/my-logstash.conf >/dev/null

    1
    chmod 755 /opt/logstash/logstash-7.7.0/runLogstash.sh

  • 运行脚本

    1
    cd /opt/logstash/logstash-7.7.0
    2
    ./runLogstash.sh

  • 查看输出文件