ELK：开源搜索与分析技术栈（2）

ELK：开源搜索与分析技术栈（2）

一、Elaticsearch Java客户端

1. Ela2

​ Elasticsearch官方提供的Java客户端分为：Java API Client（Transport）和Java Rest Client（RestAPI）两种。

1.1 Java API Client （Transport）

• 默认连接的是9300端口，基于TCP传输协议，底层基于netty框架（一种提供了NIO、AIO快速开发封装的框架）实现。
• 同版本有兼容问题。
• 通过方法调用完成与Elasticsearch的交互，有丰富的API。
• 官方自Elasticsearch 6版本开始推荐使用Java Rest Client客户端，并于7版本开始不建议使用Java API Client客户端（标记过期），并计划在8版本开始完全删除Java API Client客户端。

1.1.1 POM依赖

<dependency>
    <groupId>org.elasticsearch.clientgroupId>
    <artifactId>transportartifactId>
    <version>${elasticsearch-verion}<version>
dependency>
1
2
3
4
5

1.2 Java Rest Client

• 默认连接的是9200端口，基于http协议。
• 不同版本没有兼容问题。
• 提供了High Level和Low Level两种具体实现

1.2.1 Java Low Level Rest Client

​ 使用Apache HttpClient进行HTTP调用，只是简单封装了一下，需要自己处理请求和响应，还是面向HTTP请求的，API比较简单。

1.2.1.1 POM依赖

<dependency>
    <groupId>org.elasticsearch.clientgroupId>
    <artifactId>elasticsearch-rest-clientartifactId>
    <version>${elasticsearch-verion}version>
dependency>
1
2
3
4
5

1.2.2 Java High Level Rest Client （官方推荐使用）

​ 基于Java Low Level Rest Client是封装，提供了面向方法的API。同时请求参数和响应参数使用了elasticsearch定义的实体，方便从Java API Client迁移。并完成Elasticsearch请求响应实体转换为Java Low Level Rest Client的请求响应，即解决了Java API Client兼容问题，又解决了Java Low Level Rest Client封装使用相对复杂问题。更符合面向对象的编程思想。

1.2.2.1 POM依赖

<dependency>
    <groupId>org.elasticsearch.clientgroupId>
    <artifactId>elasticsearch-rest-high-level-clientartifactId>
    <version>${elasticsearch-verion}version>
dependency>
1
2
3
4
5

二、Spring Data Elasticsearch

​ Elasticsearch官方提供的Java客户端有其弊端，毕竟其针对的是如何访问操作Elasticsearch，而不是如何在具体的项目环境中访问操作Elasticsearch。所以官方客户端所有的API都是针对Elasticsearch中的概念，如：Index索引、Document文档等。在工作中，开发是针对具体的项目环境的，更需要一款面向对象的、简单易用的访问操作Elasticsearch的方法，这个时候就需要找一个合适的框架来替代官方客户端。Spring Data Elasticsearch框架可以满足我们的所有要求。

1. Spring Data Elasticsearch简介

​ Spring Data Elasticsearch是Spring Data框架的二级子框架，旨在为新数据存储（Elasticsearch）提供熟悉且一致的基于Spring的编程模式，同时保留特定于存储的特性和功能。其核心功能是：以POJO为中心模型，用于与Elastichsearch文档交互，并轻松编写存储库样式的数据访问层框架。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-GhYCJVYi-1663289415179)(images/es1.png)]

三、Spring Data Elasticsearch项目环境搭建

​ 既然要学习Spring Data Elasticsearch框架，那么一定学习其现阶段官方推荐使用的技术。

​ 我们学习的是底层封装了Rest High Level的ElasticsearchRestTemplate模板类型。如果工作中需要使用Java API Client（Transport），则应用ElasticsearchTemplate模板类型即可。两种类型中的方法API几乎完全一样，学会了一个，另外一个主要就是配置和类型的区别。当然了，每个不同版本的框架，在方法API上还是有些差异的。

1. 创建工程

​ 创建命名为data_es的项目。

2. POM依赖

​ Spring Data Elasticsearch在访问Elasticsearch时，是需要做Java对象和JSON格式字符串互相转换的，是基于jackson技术实现的，必须包含对应依赖，即jackson-databind。如果工程中包含spring-boot-starter-web依赖（自动包含jackson-databind），则可以省略。

<parent>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-parentartifactId>
    <version>2.3.12.RELEASEversion>
parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-data-elasticsearchartifactId>
    dependency>
    <dependency>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-testartifactId>
        <scope>testscope>
    dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.coregroupId>
        <artifactId>jackson-databindartifactId>
    dependency>
    <dependency>
        <groupId>org.projectlombokgroupId>
        <artifactId>lombokartifactId>
    dependency>
dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

3. 编辑配置文件

# 注意： 地址一定不要有末尾字符 '/'
# ElasticsearchRestTemplate客户端的配置
spring:
  elasticsearch:
    rest:
      uris: http://192.168.8.128:9200  # ES服务器所在位置。集群多节点地址用逗号分隔。默认http://localhost:9200
#  data: # ElasticsearchTemplate客户端配置。所有的方法API和ElasticsearchRestTemplate完全相同。
#    elasticsearch:
#      cluster-name: docker-cluster # ES集群的名称。已过期
#      cluster-nodes: 192.168.137.128:9300 # ES集群多节点地址。多个地址用逗号分隔。在7版本后已过期
1
2
3
4
5
6
7
8
9
10

四、基于Spring Data Elasticsearch访问Elasticsearch

1. 编辑实体类型

/**
 * Document - 描述类型和索引的映射。
 *  indexName - 索引名称
 *  shards - 主分片数量。默认值 1。
 *  replicas - 副本分片数量。默认值 1。
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Document(indexName = "stu_index", shards = 1, replicas = 0)
public class Student implements Serializable {
    /**
     * Field - 当前属性和索引中的字段映射规则。
     *  name - 字段名称。默认和当前类型属性名一致。
     *  type - 字段类型。默认使用FieldType.AUTO。自动识别。
     *  analyzer - 分词器。所有的Text类型字段默认使用standard分词器。
     *  index - 是否创建索引。默认值 true。
     *  format - 如果字段类型是date日期类型。此属性必须配置，用于设置日期格式化规则，使用DateFormat类型中的常量定义。
     */
    @Id
    @Field(name = "id", type = FieldType.Keyword)
    private Long id;
    @Field(name = "name", type = FieldType.Text, analyzer = "ik_max_word")
    private String name;
    @Field(name = "gender", type = FieldType.Keyword)
    private String gender;
    @Field(name = "age", type = FieldType.Integer)
    private int age;
    @Field(name = "hobbies", type = FieldType.Text, analyzer = "ik_smart")
    private List<String> hobbies;
    @Field(name = "isMarried", type = FieldType.Boolean)
    private boolean isMarried;
    @Field(name = "books", type = FieldType.Text, analyzer = "ik_smart", index = false)
    private String[] books;
    @Field(name = "birth", type = FieldType.Date, format = DateFormat.basic_date_time)
    private Date birth;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

2. 创建索引

​ 后续所有代码基于Spring Data Test。

@SpringBootTest
public class TestDataES {
    /**
     * Spring Data Elasticsearch客户端类型 - ElasticsearchRestTemplate客户端对象，
     *                                     由spring-boot-starter-data-elasticsearch启动时，自动装配创建。
     *  要求：
     *   1. 当前应用中，必须有启动类型，才能自动装配创建。
     *   2. 创建客户端时，根据配置文件，实现服务器连接。默认连接的ES服务器集群是：http://localhost:9200
     */
    @Autowired
    private ElasticsearchRestTemplate restTemplate;
    
    /**
     * 创建索引
     */
    @Test
    public void testCreateIndex(){
        /*
         * 在spring data elasticsearch中，所有的索引相关操作都封装在一个操作对象 IndexOperations 中。
         * 基于一个实体类型的类对象，创建索引操作对象。实体类型中，有使用注解描述的索引和映射相关信息。可以做为索引的创建的设置。
         */
        IndexOperations indexOps = restTemplate.indexOps(Student.class);

        // 创建索引
        indexOps.create();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

3. 设置映射

/**
 * 设置映射
 * 在商业开发中，几乎不会使用框架创建索引或设置映射。因为这是架构或者管理员的工作。且不适合使用代码实现
 */
@Test
public void testCreateIndexAndPutMapping(){
    IndexOperations indexOps = restTemplate.indexOps(Student.class);

    // 设置映射
    indexOps.putMapping(indexOps.createMapping());
}
1
2
3
4
5
6
7
8
9
10
11

4. 删除索引

/**
 * 删除索引
 */
@Test
public void testDeleteIndex(){
    restTemplate.indexOps(Student.class).delete();
}
1
2
3
4
5
6
7

5. 新增文档

​ 如果索引不存在，也可以执行新增文档操作，新增文档时自动创建索引。但是field通过动态mapping进行自动映射，elaticsearch根据值类型判断每个属性类型，默认每个Text字段都是standard分词器，ik分词器是不生效的。所以一定要先通过代码进行mapping设置，或直接在elasticsearch中通过命令创建所有field的mapping（推荐）。

5.1 新增单一文档

/**
 * 新增单一文档
 */
@Test
public void testAddDocument() throws Exception{
    // 创建要新增的实体对象。手工赋予主键，返回结果就是新增的数据对象。
    Student student = new Student();
    student.setId(1L);
    student.setName("张三");
    student.setGender("男");
    student.setAge(25);
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
    student.setBirth(sdf.parse("1996-01-01"));
    student.setBooks(new String[]{"Java面向对象思想","Spring In Action"});
    student.setHobbies(Arrays.asList("吃饭","上班", "睡觉"));
    student.setMarried(true);

    // 保存数据到Elasticsearch
    student = restTemplate.save(student);

    System.out.println(student);

    System.out.println("====================================================");
    
    // 自动生成主键，返回的结果中包含Elasticsearch生成的主键。
    student = new Student();
    student.setName("李四");
    student.setGender("男");
    student.setAge(21);
    student.setBirth(sdf.parse("2000-01-01"));
    student.setBooks(new String[]{"火影忍者","海贼王"});
    student.setHobbies(Arrays.asList("玩游戏","睡觉", "看动漫"));
    student.setMarried(false);

    student = restTemplate.save(student);

    System.out.println(student);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

5.2 批量新增文档

/**
 * 批量新增文档
 */
@Test
public void testBatchAdd() throws Exception {
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");

    List<Student> list = new ArrayList<>();
    list.add(new Student(2L, "郭德纲", "男", 48, Arrays.asList("郭麒麟"), true, new String[]{"德云逗笑社"}, sdf.parse("1973-01-18")));
    list.add(new Student(3L, "于谦", "男", 52, Arrays.asList("郭麒麟","抽烟","喝酒","烫头"), true, new String[]{"德云逗笑社"}, sdf.parse("1969-01-24")));

    restTemplate.save(list);

    System.out.println("批量新增结束");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

6. 删除文档

/**
 * 删除文档
 */
@Test
public void testDeleteDocument(){
    // 删除类型对应的索引中的指定主键数据，返回删除数据的主键。注意：Elasticsearch中的文档主键都是字符串类型的。
    String response = restTemplate.delete("1", Student.class);

    System.out.println(response);
}
1
2
3
4
5
6
7
8
9
10

7. 更新文档

/**
 * 更新文档。
 * save方法，可以覆盖同主键数据。案例省略
 * update更新，部分更新
 */
@Test
public void testUpdate(){
    UpdateQuery query =
            UpdateQuery.builder("2")
                    .withDocument(Document.parse("{\"hobbies\":[\"郭麒麟\", \"郭汾阳\"]}"))
                    .build();
    UpdateResponse response =
            restTemplate.update(query, IndexCoordinates.of("stu_index"));
    System.out.println(response.getResult());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

8. 主键查询文档

/**
 * 主键查询文档
 */
@Test
public void testGetById(){
    Student student = restTemplate.get("3", Student.class);
    System.out.println(student);
}
1
2
3
4
5
6
7
8

9. 全字段搜索

/**
 * querystring 搜索。所有的搜索的套路一致。具体方法如下：
 * SearchHits search(Query query, Class clazz);
 *  参数query - 具体的条件。是spring data elasticsearch框架封装的类型。
 *   一般使用构建器创建搜索条件。
 *   NativeSearchQueryBuilder().build()
 *     .withQuery(QueryBuilder) - 提供具体的条件。此方法的参数类型是QueryBuilder
 *     这个QueryBuilder类型由elasticsearch提供的java客户端jar包定义。
 *       QueryBuilder对象，一般使用构建器创建，QueryBuilders
 *  参数clazz - 返回结果每个document封装的对象类型。
 *  返回return - 搜索的结果。是一个完整的搜索结果对象。包含总数据个数，当前搜索集合，是否超时等。
 *
 *  SearchHits类型实现了Iterable接口，可迭代。容器中每个对象的类型是SearchHit。
 *  SearchHit类型中包含一个完整的搜索对象，由元数据和源数据组成。
 */
@Test
public void testQuerystring(){
    QueryBuilder queryBuilder =
            QueryBuilders.queryStringQuery("郭");

    Query query =
            new NativeSearchQueryBuilder()
                    .withQuery(queryBuilder) // 提供具体的条件。
                    .build();

    SearchHits<Student> hits =
            restTemplate.search(query, Student.class);

    System.out.println("搜索结果数据个数是： " + hits.getTotalHits());

    for(SearchHit<Student> hit : hits){
        // 源数据，就是保存在Elasticsearch中的数据
        Student content = hit.getContent();
        System.out.println("源数据：" + content);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

10. Match All搜索

/**
 * 搜索全部
 */
@Test
public void testMatchAll(){
    SearchHits<Student> hits = restTemplate.search(
            new NativeSearchQueryBuilder()
                    .withQuery(QueryBuilders.matchAllQuery())
                    .build(), Student.class);
    for(SearchHit<Student> hit : hits){
        System.out.println(hit.getContent());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

11. Match 搜索

/**
 * match搜索
 */
@Test
public void testQueryByMatch(){
    SearchHits<Student> hits = restTemplate.search(
            new NativeSearchQueryBuilder()
                    .withQuery(
                            QueryBuilders.matchQuery(
                                    "name",
                                    "于谦")
                    )
                    .build(),
            Student.class
    );

    print(hits);

}

private void print(SearchHits<Student> hits){
    for(SearchHit<Student> hit : hits){
        System.out.println(hit.getContent());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

12. Match Phrase 搜索

/**
 * match phrase搜索
 */
@Test
public void testQueryByMatchPhrase(){
    SearchHits<Student> hits = restTemplate.search(
            new NativeSearchQueryBuilder()
                    .withQuery(
                            QueryBuilders.matchPhraseQuery(
                                    "hobbies",
                                    "烫头"
                            )
                    )
                    .build(),
            Student.class
    );
    print(hits);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

13. Range 搜索

/**
 * range搜索
 */
@Test
public void testQueryByRange(){
    SearchHits<Student> hits =
            restTemplate.search(
                    new NativeSearchQueryBuilder()
                            .withQuery(
                                    QueryBuilders.rangeQuery("age")
                                    .lte(35).gte(30)
                            )
                            .build(),
                    Student.class
            );

    print(hits);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

14. Bool 搜索

/**
 * bool搜索。多条件
 */
@Test
public void testQueryByBool(){
    /*
     * bool搜索条件中包含3个条件集合。List，分别命名为must（必须匹配的条件）,mustNot（必须不匹配的条件）,should（任意匹配的条件）。
     * 每个条件集合管理方案都提供了两个方法API。以must举例如下：
     *  must() - 获取must必要条件集合，返回List。
     *  must(QueryBuilder) - 把参数添加到must条件集合中，返回BoolQueryBuilder类型对象，即当前对象本身（this）。
     */
    BoolQueryBuilder queryBuilder =
            QueryBuilders.boolQuery();
    List<QueryBuilder> must = queryBuilder.must();
    must.add(QueryBuilders.matchQuery("name", "郭"));
    queryBuilder.should(QueryBuilders.rangeQuery("age").gt(30))
            .should(QueryBuilders.rangeQuery("age").lt(60));

    // 搜索
    SearchHits<Student> hits =
            restTemplate.search(
                    new NativeSearchQueryBuilder()
                            .withQuery(queryBuilder)
                            .build(),
                    Student.class
            );

    print(hits);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

15. 分页和排序

/**
 * 分页+排序
 * 在spring data系列框架中，分页和排序可以混合处理。使用通用定义spring data commons中的Page接口实现。应用接口的实现类型PageRequest处理。
 * PageRequest类型中，提供静态方法of(int page, int size[, Sort sort]);
 * page - 查询第几页，页码数字从0开始计数。
 * size - 查询多少行。
 * Sort - 具体的排序规则。可选参数。
 */
@Test
public void testQueryByPageAndSort(){
    SearchHits<Student> hits =
            restTemplate.search(
                    new NativeSearchQueryBuilder()
                            .withQuery(QueryBuilders.matchAllQuery())
                            .withPageable(
                                    PageRequest.of(0, 2,
                                            Sort.by(
                                                    Sort.Order.desc("age"),
                                                    Sort.Order.asc("id")
                                            )
                                    )
                            )
                            .build(),
                    Student.class
            );
    print(hits);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

16. 高亮处理

/**
 * 搜索结果高亮处理
 * 提供高亮字段，字段名，是否分段（每段几个字符，返回几段），前后缀。
 * 创建Query对象，做搜索处理。
 */
@Test
public void testQueryHighLight(){
    // 创建高亮字段，必须和具体的字段名绑定。
    HighlightBuilder.Field field1 = new HighlightBuilder.Field("name");
    // 高亮前缀
    field1.preTags("");
    // 高亮后缀
    field1.postTags("");
    // 分段的每段字符数
    field1.fragmentSize(Integer.MAX_VALUE);
    // 分段后，返回几段
    field1.numOfFragments(1);
    Query query =
            new NativeSearchQueryBuilder()
                    .withQuery(
                            QueryBuilders.matchQuery(
                                    "name",
                                    "于谦")
                    )
                    .withHighlightFields(field1)
                    .build();

    SearchHits<Student> hits =
            restTemplate.search(query, Student.class);

    for (SearchHit<Student> hit : hits){
        // 获取源数据
        Student content = hit.getContent();
        // 找高亮数据
        List<String> hl = hit.getHighlightField("name");
        // 判断是否有高亮数据
        if(hl != null && hl.size() > 0){
            // 有高亮数据
            content.setName(hl.get(0));
        }
        System.out.println(content);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

五、LogStash简介

1. 什么是LogStash

​ ELK(Elasticsearch+Logstash+Kibana)中我们使用过Elasticsearch和Kibana，就剩下最后一个LogStash了。
​ 到底Logstash是什么呢？官方说明：Logstash 是开源的服务器端数据处理管道，能够同时从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的“存储库”中。

​ 通俗说明：Logstash是一款强大的数据处理工具，常用作日志处理。
​ 到目前为止，Logstash已经有超过200个可用的插件，以及创建和贡献自己的灵活性。社区生态非常完善，对于我们可以放心的使用。

2. 为什么使用Logstash

​ 通常当系统发生故障时，工程师需要登录到各个服务器上，使用 grep / sed / awk 等 Linux 脚本工具去日志文件里查找故障原因。在没有日志系统的情况下，首先需要定位处理请求的服务器，如果这台服务器部署了多个实例，则需要去每个应用实例的日志目录下去找日志文件。每个应用实例还会设置日志滚动策略（如：每天生成一个文件），还有日志压缩归档策略等。

​ 这样一系列流程下来，对于我们排查故障以及及时找到故障原因，造成了比较大的麻烦。因此，如果我们能把这些日志集中管理，并提供集中检索功能，不仅可以提高诊断的效率，同时对系统情况有个全面的理解，避免事后救火的被动。

​ 所以日志集中管理功能就可以使用ELK技术栈进行实现。Elasticsearch具有数据存储和分析的能力，Kibana提供可视化管理平台。缺少的数据收集和整理的角色，就是由Logstash负责的。

3. Logstash工作原理（面试题）

3.1 Data Source 数据源

​ Logstash 支持的数据源有很多。例如对于日志功能来说只要有日志记录和日志传递功能的系统都支持。
​ Spring Boot中默认推荐logback支持日志输出功能，可以输出日志到控制台、数据库、文件、远程服务等。我们工作中一般会使用logback进行日志管理，可以基于Logstash插件，输出到远程Logstash服务器。

3.2 Logstash Pipeline 管道

​ Pipeline整体就是Logstash的核心。其由Input、Filter、Output三个组件组成。

3.2.1 Input

​ 输入源，一般配置为Logstash监听的主机及端口。DataSource向指定的ip及端口（Logstash）输出日志，Input 输入源监听到数据信息就可以进行收集。

3.2.2 Filter

​ 过滤器，对收集到的信息进行过滤（额外处理），如果不需额外处理，也可以省略这个组件配置。

3.2.3 Output

​ 把收集到的信息发送给谁。在ELK技术栈中都是输出到Elasticsearch中，后面数据检索和数据分析的过程也是由Elasticsearch实现的。

​ 最终效果：通过整体步骤可以把原来一行日志信息转换为Elasticsearch支持的Document数据存储到Elasticsearch中。

六、基于Docker安装Logstash

1. 拉取镜像

docker pull logstash:7.6.2
1

2. 创建并启动容器

​ 端口号映射问题：Logstash没有默认监听端口。监听端口取决于配置中的Input输入组件监听的端口。案例中提前定义4560端口映射，在后续配置中，会在Input组件中监听此端口。

docker run  -p 4560:4560 --name logstash -d  logstash:7.6.2
1

3. 配置Logstash Pipeline

3.1 进入容器

docker exec -it logstash /bin/bash
1

3.2 修改logstash.yml核心配置

vi /usr/share/logstash/config/logstash.yml
1

3.2.1 配置内容修改如下：

​ 如果Elasticsearch为集群，多个地址使用数组方式进行配置。

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://192.168.8.128" ]
1
2

3.3 修改Pipeline管道配置文件

vi /usr/share/logstash/pipeline/logstash.conf
1

3.3.1 配置内容修改如下：

input {
        tcp {
                mode => "server"
                port => 4560
        }
}
filter {
}
output {
        elasticsearch {
                action => "index"
                hosts  => "192.168.8.128:9200"
                index  => "log_index"
        }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3.3.2配置含义：

input { # 输入组件配置。
        tcp { # 基于TCP协议提供监听服务。
                mode => "server" # 模式为服务器模式。
                port => 4560 # 监听4560端口。
        }
}
filter { # 过滤器组件配置，因没有额外功能，可省略，也可空配置。
}
output { # 输出组件配置。
        elasticsearch { # 输出到Elasticsearch服务器。
                action => "index" # 输出到Elasticsearch时使用的命令，index就是保存，主键存在则覆盖，主键不存在则新增。
                hosts  => "192.168.8.128:9200" # Elasticsearch服务器地址。
                index  => "log_index" # 保存日志文档的索引名，此索引不存在则创建默认规则索引。
        }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

4. 重启容器

​ 推出容器并重新启动容器

exit

docker restart logstash
1
2
3

5. 查看启动日志

​ 在启动日志中，看到Successful Started即为启动成功。

七、使用Logback向Logstash中输出日志

​ 需求：基于Logback技术，把项目的日志输出到控制台，同时也输出到Logstash中。

1. 创建工程

​ 创建命名为 my_app 的工程。

2. POM依赖

<parent>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-parentartifactId>
    <version>2.3.12.RELEASEversion>
parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-webartifactId>
    dependency>
    <dependency>
        <groupId>net.logstash.logbackgroupId>
        <artifactId>logstash-logback-encoderartifactId>
        <version>6.3version>
    dependency>
    <dependency>
        <groupId>org.projectlombokgroupId>
        <artifactId>lombokartifactId>
    dependency>
dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

3. 编辑配置文件

​ 编辑applicaiton.yml配置文件

server:
  port: 8888
spring:
  application:
    name: my-app
1
2
3
4
5

4. 编辑logback配置文件

​ 在src/main/resources/目录中，编辑logback.xml配置文件。此配置文件在今天的软件目录中已提供。

<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml" />

    <springProperty scope="context" name="springAppName"
                    source="spring.application.name" />

    
    <property name="LOG_FILE" value="${BUILD_FOLDER:-build}/${springAppName}" />

    
    <property name="CONSOLE_LOG_PATTERN"
              value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}" />

    
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>INFOlevel>
        filter>
        
        <encoder>
            <pattern>${CONSOLE_LOG_PATTERN}pattern>
            <charset>utf8charset>
        encoder>
    appender>
    
    <appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>192.168.8.128:4560destination>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder" />
    appender>
    
    <root level="DEBUG">
        <appender-ref ref="console" />
        <appender-ref ref="logstash" />
    root>
configuration>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

5. 编辑控制器

/**
 * Log注解 - lombok提供的注解。在当前类型中，提供一个可以做日志处理的属性。属性名称是 log。 类型是Logger。其基于spring框架默认依赖的日志插件框架logback实现。
 * log中的方法：
 *  info - 信息日志
 *  warning - 警告日志
 */
@RestController
@Log
public class MyController {
    @RequestMapping("/test")
    public String test(){
        try {
            System.out.println("测试成功");
            log.info("测试成功");
            return "测试成功";
        }catch (Exception e){
            System.out.println("测试失败");
            log.warning("测试失败");
            return "测试失败";
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

6. 编辑启动类型

@SpringBootApplication
public class MyApp {
    public static void main(String[] args) {
        SpringApplication.run(MyApp.class, args);
    }
}
1
2
3
4
5
6

7. 启动

​ 启动后，访问 http://localhost:8888/test 查看控制台日志。

八、基于Kibana仪表盘查看日志信息

1. 使用命令查看日志信息

​ 在Kibana中输入如下命令，查看日志信息：

GET /test_log/_search
{
  "query": {
    "match": {
      "message": "测试成功"
    }
  }
}
1
2
3
4
5
6
7
8

​ 查询结果如下:

2. 基于图像界面查看

2.1 进入Kibana Index Pattern管理面板

2.2 新建Index Pattern

2.3 选择时间过滤并创建Index Pattern

2.4 基于Discover查看索引数据

3. 索引中文档结构解释

​ 文档数据来源于Java应用中的日志系统logback。具体如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-UqcAkbfT-1663289415206)(images/es14.png)]

​ logback日志经由logstash-logback-encoder插件，转换日志数据，并发送到远程服务器Logstash的Input组件中，经由Filter组件过滤处理（本案例中没有额外处理逻辑，已省略Filter组件配置），最终经由Output组件输出到远程服务器Elasticsearch的索引test_log中，保存为文档。形成Elasticsearch中的文档数据。

​ 索引中的文档各字段含义如下：

九、基于ELK搭建日志查询系统

​ 绝大多数项目在后台管理中都有日志管理系统。以前的日志信息是存储在MySQL中，日志随着项目的运行会越来越多，一直存储在MySQL会导致查询性能降低。现在的日志信息通过ELK技术栈进行管理。存储在Elasticsearch的索引中，可以更好的分析日志内容、提供更快搜索效率和更好的搜索召回率。
​ 在这里，我们开发一个简易日志管理系统，给定需求如下：

• 搭建日志系统，提供查询Elasticsearch中日志信息的搜索接口。

1. 新建工程

​ 新建命名为log_manager的工程

2. POM依赖

<parent>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-parentartifactId>
    <version>2.3.12.RELEASEversion>
parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-webartifactId>
    dependency>
    <dependency>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-thymeleafartifactId>
    dependency>
    <dependency>
        <groupId>org.projectlombokgroupId>
        <artifactId>lombokartifactId>
    dependency>
    <dependency>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-data-elasticsearchartifactId>
    dependency>
dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

3. 编辑配置文件

​ 编辑application.yml配置文件

server:
  port: 9999
spring:
  application:
    name: log-management
  elasticsearch:
    rest:
      uris: http://192.168.8.128:9200
1
2
3
4
5
6
7
8

4. 编辑实体类型

4.1 编辑对应索引的实体类型LogMessage

/**
 * 日志实体类型。对应Elasticsearch中的索引test_log
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Document(indexName = "test_log")
public class LogMessage implements Serializable {
    @Id
    private String id;
    @Field(name = "@timestamp", type = FieldType.Date, format = DateFormat.date_time)
    private Date timestamp;
    @Field(name = "@version", type = FieldType.Text)
    private String version;
    @Field(name = "host", type = FieldType.Text)
    private String host;
    @Field(name = "message", type = FieldType.Text)
    private String message;
    @Field(name = "port", type = FieldType.Long)
    private String port;
    private Message messageObject;

    /**
     * 提供通过JSON格式字符串转换成Java对象的方法。
     * @param messageJSON
     */
    public void json2Obj(String messageJSON) {
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            this.messageObject = objectMapper.readValue(messageJSON, Message.class);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            this.messageObject = new Message();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

4.2 编辑对应JSON格式字段message的实体类型Message

/**
 * 日志消息实体类型。对应Elasticsearch中的索引test_log中的message字段内容
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Message implements Serializable {
    @JsonProperty("@timestamp")
    private Date timestamp;
    @JsonProperty("@version")
    private String version;
    private String message;
    @JsonProperty("logger_name")
    private String loggerName;
    @JsonProperty("thread_name")
    private String threadName;
    private String level;
    @JsonProperty("level_value")
    private String levelValue;
    private String springAppName;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

5. 编辑数据访问接口

/**
 * 日志数据访问接口
 */
public interface LogMessageDao {
    /**
     * 搜索
     * @param query 搜索条件
     * @param page 第几页，从0开始
     * @param size 多少行
     * @return 返回结果使用Map集合，给定两组键值对，含义如下：
     *  key             value
     *  total           总计数据行数  long类型
     *  contents        当前页显示的数据集合 List
     */
    Map<String, Object> search(String query, int page, int size);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

6. 编辑数据访问实现

/**
 * 数据访问实现类
 */
@Repository
public class LogMessageDaoImpl implements LogMessageDao {
    @Autowired
    private ElasticsearchRestTemplate restTemplate;

    /**
     * 搜索
     *
     * @param query 搜索条件，query为null或者空字符串，则全搜索。无高亮
     *              不是null，则条件搜索，条件为message属性。高亮处理。
     * @param page 第几页，从0开始
     * @param size 多少行
     * @return
     */
    @Override
    public Map<String, Object> search(String query, int page, int size) {

        Query q = null;

        if (query == null || query.trim().length() == 0){
            // 无条件搜索
            q = new NativeSearchQueryBuilder()
                    .withQuery(QueryBuilders.matchAllQuery())
                    .withPageable(PageRequest.of(page, size))
                    .build();
        }else{
            // 有条件搜索
            HighlightBuilder.Field f = new HighlightBuilder.Field("message");
            f.fragmentSize(Integer.MAX_VALUE);
            f.numOfFragments(1);
            f.preTags("");
            f.postTags("");
            f.highlighterType("plain");
            q = new NativeSearchQueryBuilder()
                    .withQuery(
                            QueryBuilders.matchQuery("message", query)
                    )
                    .withHighlightFields(f)
                    .withPageable(PageRequest.of(page, size))
                    .build();
        }

        // 搜索
        SearchHits<LogMessage> hits =
                restTemplate.search(
                        q,
                        LogMessage.class
                );

        // 处理结果
        Map<String, Object> result = new HashMap<>();
        result.put("total", hits.getTotalHits()); // 总计数据行数

        // 处理搜索结果
        List<LogMessage> list =
                new ArrayList<>();
        for(SearchHit<LogMessage> hit : hits){
            LogMessage message = hit.getContent();
            // 处理高亮
            List<String> hl = hit.getHighlightField("message");
            if(hl != null && hl.size() > 0){
                // 有高亮数据
                message.setMessage(hl.get(0));
            }
            // 把处理后的日志对象，保存到集合中
            list.add(message);
        }

        result.put("contents", list);

        return result;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

7. 编辑服务接口

public interface LogMessageService {
    /**
     * 分页搜索日志
     * @param q 搜索条件
     * @param page 搜索第几页
     * @param size 每页显示多少行数据
     * @return
     */
    Map<String, Object> search(String q, int page, int size);
}
1
2
3
4
5
6
7
8
9
10

8. 编辑服务实现

@Service
public class LogMessageServiceImpl implements LogMessageService {
    @Autowired
    private LogMessageDao dao;

    /**
     * 搜索
     * @param q
     * @param page
     * @param size
     * @return
     */
    @Override
    public Map<String, Object> search(String q, int page, int size) {
        Map<String, Object> result = dao.search(q, page, size);

        // 把搜索结果中JSON格式的message字符串，转换成Message类型的对象
        List<LogMessage> list = (List<LogMessage>) result.get("contents");
        for(LogMessage logMessage : list){
            logMessage.json2Obj(logMessage.getMessage());
        }

        // 设置当前页码
        result.put("currPage", page);
        // 获取总计数据行数
        long total = (long) result.get("total");
        long pages = total % size == 0 ? ((total / size) - 1) : (total / size);
        // 设置最大页码
        result.put("pages", pages);
        // 设置搜索条件
        result.put("q", q);
        result.put("size", size);

        return result;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

9. 编辑控制器

@Controller
public class LogMessageController {

    @Autowired
    private LogMessageService service;

    @RequestMapping(value = {"/search", "/"})
    public String search(String q,
                         @RequestParam(name = "page", defaultValue = "0") int page,
                         @RequestParam(name = "size", defaultValue = "5") int size,
                         Model model){
        model.addAllAttributes(service.search(q, page, size));
        return "logManager";
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

10. 编辑HTML页面

​ 在src/main/resources/templates文件夹中编辑logManager.html文件：

DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>日志管理系统title>
head>
<body>
<div style="margin: auto; width: 800px; ">
    <form action="/" method="get">
        <input name="q" value="">
        <input type="submit" value="搜索">
    form>
    <table style="width:800px" border="1">
        <thead>
        <tr>
            <th>主键th>
            <th>主机th>
            <th>端口th>
            <th>版本th>
            <th>时间th>
            <th>日志级别th>
            <th>类型名称th>
            <th>线程名称th>
            <th>消息th>
        tr>
        thead>
        <tbody>
        <tr th:each="log : ${contents}">
            <td th:text="${log.id}">td>
            <td th:text="${log.host}">td>
            <td th:text="${log.port}">td>
            <td th:text="${log.version}">td>
            <td th:text="${log.timestamp}">td>
            <td th:utext="${log.messageObject.level}">td>
            <td th:utext="${log.messageObject.loggerName}">td>
            <td th:utext="${log.messageObject.threadName}">td>
            <td th:utext="${log.messageObject.message}">td>
        tr>
        <tr>
           <td colspan="9" style="text-align: center">
               <a th:if="${currPage > 0}" th:href="@{/(q=${q},page=${currPage - 1},size=${size})}">上一页a>
               <a th:if="${currPage < pages}" th:href="@{/(q=${q},page=${currPage + 1},size=${size})}">下一页a>
               <span th:if="${currPage == 0 && currPage == pages}" >没有更多数据了span>
           td>
        tr>
        tbody>
    table>

div>
body>
html>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

11. 编辑启动类型

@SpringBootApplication
public class LogManagementApp {
    public static void main(String[] args) {
        SpringApplication.run(LogManagementApp.class, args);
    }
}
1
2
3
4
5
6

12. 启动测试

​ 启动应用，访问 http://localhost:9999/ ，显示默认搜索结果如下：

​ 在搜索条件输入框中输入 “测试成功” ，搜索结果如下：