1、前期准备

在实现日志采集前，需要搭建好Logstash环境，比如环境情况如下：

• 解压安装目录：/home/logstash/logstash-7.8.0
• 启动加载的配置目录：/home/logstash/logstash-7.8.0/config/logs
• 启动命令：/home/logstash/logstash-7.8.0/bin/logstash -f /home/logstash/logstash-7.8.0/config/logs

2、引入POM

<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>7.2</version>
</dependency>

3、配置logback.xml

在原有的的logback.xml配置文件中，加上LogstashEncoder配置，部分配置如下：

关键配置为LogstashTcpSocketAppender中的内容，<destination>表示连接logstash的IP及端口地址；

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
    
    <property resource="application.yml"/>
    <!--引入项目名称作为标识-->
    <springProperty scope="context" name="application.name" source="spring.application.name"/>
    <!-- 日志存放路径 -->
    <property name="log.path" value="logs"/>
    <!-- 日志输出格式 -->
    <property name="log.pattern"
              value="%clr(%d{${yyyy-MM-dd HH:mm:ss.SSS}}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(-&#45;&#45;){faint} %clr([%20.20t]){faint} %clr(%-40.40logger{39}){cyan} : %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}"/>

    <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
		<!-- logstash的IP及logstash中input配置的端口 -->
        <destination>192.168.80.121:9092</destination>
        <includeCallerData>true</includeCallerData>

        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
            <includeCallerData>true</includeCallerData>
        </encoder>
        <encoder>
            <pattern>${log.pattern}</pattern>
            <!-- 记录日志的编码:此处设置字符集 - -->
            <charset>UTF-8</charset>
        </encoder>
    </appender>

    <!-- 控制台输出 -->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>${log.pattern}</pattern>
            <!-- 记录日志的编码:此处设置字符集 - -->
            <charset>UTF-8</charset>
        </encoder>
    </appender>

    <root level="info">
        <appender-ref ref="console"/>
    </root>

    <!--系统操作日志-->
    <root level="info">
        <appender-ref ref="LOGSTASH"/>
    </root>
</configuration>

4、修改logstash配置

对于日志采集的相关配置文件，都在home/logstash/logstash-7.8.0/config/logs目录下；

新增配置文件logstash-logs.conf，配置内容如下：

input {
    tcp {
        mode => "server"
        host => "0.0.0.0"
        port => 9092
        codec => json_lines
    }
}

output {
       file {
       path => "/home/logs/%{+YYYY-MM}/%{+YYYY-MM-dd}/%{+YYYY-MM-dd-HH}/application-%{+HH}.log"
       codec => line
    }
}

配置说明：

9092：与logback.xml文件中配置<destination>标签指定的端口一致；

application：与logback.xml文件中<customFields>标签指定的变量名称一致，用于区分不同的服务名称；

输出配置：

可以将采集的日志输出到es、hdfs、file等目的地

elasticsearch{
    hosts=>["192.168.80.121:9200","192.168.80.122:9200"]
    index=>"logs-%{+YYYY-MM-dd}"
}
webhdfs {
    workers => 2
    host => "192.168.80.121"
    port => "50070"
    user => "hdfs"
    path => "/logstash-%{+YYYY-MM-dd}.log"     # 按天创建log
    codec => "json"
}
file {
	path => "/home/logs/%{+YYYY-MM}/%{+YYYY-MM-dd}/%{+YYYY-MM-dd-HH}/application-%{+HH}.log"
	codec => line
}

多服务处理模式：

如果存在多个服务同时往同一个logstash中存放日志信息，可以采用两种方式进行实现

方式一：

建立多个logstash-xx.conf文件，每个文件里面的input的端口号以及output的输出路径不同即可；

方式二：

在同一个logstash-logs.conf文件中，配置多个input及output，比如：

input{
 tcp {
		mode => "server"
     host => "0.0.0.0"
     port => 9092
     codec => json_lines
	    type => "order"
 }
}
input{
 tcp {
		mode => "server"
     host => "0.0.0.0"
     port => 9093
     codec => json_lines
	    type => "product"
 }
}

filter {
	if[type] == "order" {
		mutate {
	  		add_tag => ["order"]
	  	}
	}

	if [type] == "product" {
		mutate {
			add_tag => ["product"]
		}
	} 
}


output {
	if "order" in [tags] {
     file {
        path => "/home/logs/order/%{+YYYY-MM}/%{+YYYY-MM-dd}/%{+YYYY-MM-dd-HH}/application-%{+HH}.log"
        codec => line 
     }
	}

	if "product" in [tags] {
     file {
        path => "/home/logs/product/%{+YYYY-MM}/%{+YYYY-MM-dd}/%{+YYYY-MM-dd-HH}/application-%{+HH}.log"
        codec => line  
     }	
	}	
}

5、采集测试

如果在日志采集过程中，logstash服务挂掉了，不会影响应用程序的正常运行，只是无法继续采集日志

1、启动logstah

# 启动命令：
/home/logstash/logstash-7.8.0/bin/logstash -f /home/logstash/logstash-7.8.0/config/logs

2、启动项目

如果logstash搭建成功，在启动项目后，会在output配置的路径中生成日志文件，并且打印日志信息，如下：