本文基于之前搭建的ELK平台进行实时及离线日志采集的实例开发。
实时日志采集通过log4j的socket方式发送日志到logstash，离线日志采集通过logstash的shipper角色对日志文件进行扫描。

1.准实时日志分析测试

1.1建立Java工程elk-logs-producer

Application.java

public class Application {

    private static final Logger LOGGER = Logger.getLogger(Application.class);
    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 50; i++) {
            LOGGER.debug("debug log [" + i + "]. 这是一条调试日志!");
            LOGGER.info("Info log [" + i + "]. 这是一条信息日志!");
            LOGGER.warn("warn log [" + i + "]. 这是一条告警日志!");
            LOGGER.error("error log [" + i + "]. 这是一条错误日志!");
            Thread.sleep(500);
        }
    }
}

Log4j.properties

log4j.appender.logstash=org.apache.log4j.net.SocketAppender
log4j.appender.logstash.RemoteHost=127.0.0.1
log4j.appender.logstash.port=4560
log4j.appender.logstash.ReconnectionDelay=60000
log4j.appender.logstash.LocationInfo=true

Pom.xml

<project
xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.snowalker</groupId>
  <artifactId>elk-logs-producer</artifactId>
  <version>0.0.1-SNAPSHOT</version>

  <dependencies>
      <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
  </dependencies>
</project>

1.2运行Java工程，观察logstash控制台信息

通过kibana搜索15分钟内日志

2.离线日志分析

2.1 配置logstash.conf如下

input {
    file {
        path => "D:/apache-tomcat-8.0.30/logs/*"
    }
}

filter {

}

output {
    stdout {}
    elasticsearch{
        hosts => ["localhost:9200"]
        index => "log4j-%{+YYYY.MM.dd}"
        document_type => "file_log"
    }
}

解释：

扫描tomcat安装路径下的logs文件
输出到logstash控制台以及输入到elasticsearch中。

2.2使用新配置文件启动logstash

查看输出的结果

在kibana中搜索关键字

小结

我们进行日志分析的步骤一般如下图

1. 将logstash或者filebeat等日志搜集工具(shipper)部署到具体的应用所在的环境中扫描对应的工程的日志
2. 将这些日志集中写入到消息队列(broker)中
3. logstash对队列中的信息进行统一的索引(Indexer)
4. 消息队列将数据写入到es集群中
5. 在kibana中设定具体的查询过滤关键字进行日志的搜索