今天给各位分享快速搭建图书检索服务：Spring Boot、Elasticsearch 和 Logstash的知识，其中也会对进行解释，如果能碰巧解决你现在面临的问题，别忘了关注本站，现在开始吧！

前面我们介绍了Spring Boot 整合 Elasticsearch 实现数据查询检索的功能，在实际项目中，我们的数据一般存储在数据库中，而且随着业务的发送，数据也会随时变化。

那么如何保证数据库中的数据与Elasticsearch存储的索引数据保持一致呢? 最原始的方案就是：当数据发生增删改操作时同步更新Elasticsearch。但是这样的设计耦合太高。接下来我们介绍一种非常简单的数据同步方式：Logstash 数据同步。

一、Logstash简介

1.什么是Logstash

logstash是一个开源的服务器端数据处理工具。简单来说，就是一根具备实时数据传输能力的管道，负责将数据信息从管道的输入端传输到管道的输出端;与此同时这根管道还可以让你根据自己的需求在中间加上滤网，Logstash提供里很多功能强大的滤网以满足你的各种应用场景。

Logstash常用于日志系统中做日志采集设备，最常用于ELK中作为日志收集器使用。

2.Logstash的架构原理

Logstash的基本流程架构：input=》 filter =》 output 。

filter(过滤器)负责数据处理与转换。主要是将event通过output发出之前对其实现的某些处理功能。

output(输出)：将我们过滤出的数据保存到那些数据库和相关存储中，。

 

 

3.Logstash如何与Elasticsearch数据同步

实际项目中，我们不可能通过手动添加的方式将数据插入索引库，所以需要借助第三方工具，将数据库的数据同步到索引库。此时，Logstash出现了，它可以将不同数据库的数据同步到Elasticsearch中。保证数据库与Elasticsearch的数据保持一致。

 

 

目前支持数据库与ES数据同步的插件有很多，个人认为Logstash是众多同步mysql数据到es的插件中，最稳定并且最容易配置的一个。

二、安装Logstash

Logstash的使用方法也很简单，下面讲解一下，Logstash是如何使用的。需要说明的是：这里以windows 环境为例，演示Logstash的安装和配置。

1.下载Logstash

首先，下载对应版本的Logstash包，可以通过上面提供下载elasticsearch的地址进行下载，完成后解压。

 

 

上面是Logstash解压后的目录，我们需要关注是bin目录中的执行文件和config中的配置文件。一般生产情况下，会使用Linux服务器，并且会将Logstash配置成自启动的服务。这里测试的话，直接启动。

2.配置Logstash

接下来，配置Logstash。需要我们编写配置文件，根据官网和网上提供的配置文件，将其进行修改。

第一步：在Logstash根目录下创建mysql文件夹，添加mysql.conf配置文件，配置Logstash需要的相应信息，具体配置如下：

input { 
    stdin { 
    } 
    jdbc { 
      # mysql数据库连接 
      jdbc_connection_string => "jdbc:mysql://localhost:3306/book_test?characterEncoding=utf-8&useSSL=false&serverTimezone=UTC" 
      # mysqly用户名和密码 
      jdbc_user => "root" 
      jdbc_password => "root" 
      # 驱动配置 
      jdbc_driver_library => "C:\Users\Administrator\Desktop\logstash-7.5.1\mysql\mysql-connector-java-8.0.20.jar" 
      # 驱动类名 
      jdbc_driver_class => "com.mysql.cj.jdbc.Driver" 
      #jdbc_paging_enabled => "true" 
      #jdbc_page_size => "50000" 
      jdbc_default_timezone => "Asia/Shanghai" 
      # 执行指定的sql文件 
      statement_filepath => "C:\Users\Administrator\Desktop\logstash-7.5.1\mysql\sql\bookquery.sql" 
      use_column_value => true 
      # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）； 
      lowercase_column_names => false 
      # 需要记录的字段，用于增量同步，需是数据库字段 
      tracking_column => updatetime 
      # Value can be any of: numeric,timestamp，Default value is "numeric" 
      tracking_column_type => timestamp 
      # record_last_run上次数据存放位置； 
      record_last_run => true 
      #上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值 
      last_run_metadata_path => "C:\Users\Administrator\Desktop\logstash-7.5.1\mysql\sql\logstash_default_last_time.log" 
      # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false； 
      clean_run => false 
      # 设置监听 各字段含义 分 时 天 月  年 ，默认全部为*代表含义：每分钟都更新 
      schedule => "* * * * *" 
      # 索引类型 
      type => "id" 
    } 
} 
output { 
    elasticsearch { 
        #es服务器 
        hosts => ["10.2.1.231:9200"] 
        #ES索引名称 
        index => "book" 
        #自增ID 
        document_id => "%{id}" 
    } 
     
    stdout { 
        codec => json_lines 
    } 
} 

第二步：将mysql-connector-java.jar 拷贝到前面配置的目录下。上面的mysql.conf配置的是：C:\Users\Administrator\Desktop\logstash-7.5.1\mysql\

mysql-connector-java-8.0.20.jar。那么jar包拷贝到此目录下即可：

 

 

上面是mysql的驱动，如果是sqlserver数据库，下载SqlServer对应的驱动即可。放置的位置要与mysql.conf 配置文件中的jdbc_driver_library 地址保持一致。

第三步：创建sql目录，创建bookquery.sql文件用于保存需要执行的sql 脚本。示例代码如下：

select * from book where updatetime >= :sql_last_value 
order by updatetime desc 

这里使用的增量更新，所以使用:sql_last_value 记录上一次记录的最后时间。

3.启动Logstash

进入logstash的bin目录，执行如下命令：

logstash.bat -f C:\Users\Administrator\Desktop\logstash-7.5.1\mysql\mysql.conf 

启动成功之后，Logstash就会自动定时将数据写入到Elasticsearch。如下图所示：

 

 

同步完成后，我们使用Postman查询Elasticsearch，验证索引是否都创建成功。在postman中，发送 Get 请求：

http://10.2.1.231:9200/book/_search 。返回结果如下图所示：

 

 

可以看到，数据库中的数据已经通过Logstash同步至Elasticsearch。说明Logstash配置成功。

三、创建查询服务

数据同步完成后，接下来我们使用Spring Boot 构建Elasticsearch查询服务。首先创建Spring Boot项目并整合Elasticsearch，这个之前都已经介绍过，不清楚的朋友可以看我之前的文章。

接下来演示如何封装完整的数据查询服务。

1.数据实体

@Document( indexName = "book" , replicas = 0) 
public class Book { 
    @Id 
    private Long id; 
    @Field(analyzer = "ik_max_word",type = FieldType.Text) 
    private String bookName; 
    @Field(analyzer = "ik_max_word",type = FieldType.Text) 
    private String author; 
    private float price; 
    private int page; 
    @Field(type = FieldType.Date,format = DateFormat.custom,pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'") 
    private Date createTime; 
    @Field(type = FieldType.Date,format = DateFormat.custom,pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'") 
    private Date updateTime; 
    @Field(analyzer = "ik_max_word",type = FieldType.Text) 
    private String category; 
    public Long getId() { 
        return id; 
    } 
    public void setId(Long id) { 
        this.id = id; 
    } 
    public String getBookName() { 
        return bookName; 
    } 
    public void setBookName(String bookName) { 
        this.bookName = bookName; 
    } 
    public String getAuthor() { 
        return author; 
    } 
    public void setAuthor(String author) { 
        this.author = author; 
    } 
    public float getPrice() { 
        return price; 
    } 
    public void setPrice(float price) { 
        this.price = price; 
    } 
    public int getPage() { 
        return page; 
    } 
    public void setPage(int page) { 
        this.page = page; 
    } 
    public String getCategory() { 
        return category; 
    } 
    public void setCategory(String category) { 
        this.category = category; 
    } 
    public Book(){ 
    } 
    public Date getCreateTime() { 
        return createTime; 
    } 
    public void setCreateTime(Date createTime) { 
        this.createTime = createTime; 
    } 
    public Date getUpdateTime() { 
        return updateTime; 
    } 
    public void setUpdateTime(Date updateTime) { 
        this.updateTime = updateTime; 
    } 
} 

2.请求封装类

public class BookQuery { 
    public String category; 
    public String bookName; 
    public String author; 
    public int priceMin; 
    public int priceMax; 
    public int pageMin; 
    public int pageMax; 
    public String sort; 
    public String sortType; 
    public int page; 
    public int limit; 
} 

3.创建Controller控制器

@RestController 
public class ElasticSearchController { 
    @Autowired 
    private ElasticsearchRestTemplate elasticsearchRestTemplate; 
    /** 
     * 查询信息 
     * @param 
     * @return 
     */ 
    @PostMapping(value = "/book/query") 
    public JSONResult query(@RequestBody BookQuery bookQuery){ 
        Query query= getQueryBuilder(bookQuery); 
        SearchHits searchHits = elasticsearchRestTemplate.search(query, Book.class); 
        List> result = searchHits.getSearchHits(); 
        return JSONResult.ok(result); 
    } 
    public Query getQueryBuilder(BookQuery query) { 
               BoolQueryBuilder builder = boolQuery(); 
        // 匹配器 模糊查询部分，分析器使用ik (ik_max_word) 
        List must = builder.must(); 
        if (query.getBookName()!=null && !query.getBookName().isEmpty()) 
            must.add(wildcardQuery("bookName", "*" +query.getBookName()+ "*")); 
        if (query.getCategory()!=null && !query.getCategory().isEmpty()) 
            must.add(wildcardQuery("category", "*" +query.getCategory()+ "*")); 
        if (query.getAuthor()!=null && !query.getAuthor().isEmpty()) 
            must.add(wildcardQuery("author", "*" +query.getAuthor()+ "*")); 
        // 筛选器 精确查询部分 
        List filter = builder.filter(); 
        // 范围查询 
        if (query.getPriceMin()>0 && query.getPriceMax()>0) { 
            RangeQueryBuilder price = rangeQuery("price").gte(query.getPriceMin()).lte(query.getPriceMax()); 
            filter.add(price); 
        } 
        // 范围查询 
        if (query.getPageMin()>0 && query.getPageMax()>0) { 
            RangeQueryBuilder page = rangeQuery("page").gte(query.getPageMin()).lte(query.getPageMax()); 
            filter.add(page); 
        } 
        // 分页 
        PageRequest pageable = PageRequest.of(query.getPage() - 1, query.getLimit()); 
        // 排序 
        SortBuilder sort = SortBuilders.fieldSort("price").order(SortOrder.DESC); 
        //设置高亮效果 
        String preTag = "";//google的色值 
        String postTag = ""; 
        HighlightBuilder.Field highlightFields = new HighlightBuilder.Field("category").preTags(preTag).postTags(postTag); 
        Query searchQuery = new NativeSearchQueryBuilder() 
                .withQuery(builder) 
                .withHighlightFields(highlightFields) 
                .withPageable(pageable) 
                .withSort(sort) 
                .build(); 
        return searchQuery; 
    } 
} 

4.测试验证

启动项目，在Postman中，请求

http://localhost:8080/book/query 接口查询书籍信息数据。查看接口返回情况。

 

 

我们看到接口成功返回数据。说明数据查询服务创建成功。

最后

以上，我们就把使用Spring Boot + Elasticsearch + Logstash 实现完整的数据查询检索服务介绍完了。

 

END，本文到此结束，如果可以帮助到大家，还望关注本站哦！

用户评论

断桥残雪

这个系列博客看起来很有实用性！想要了解一下Spring Boot,Elasticsearch 和Logstash 的应用场景。

    有10位网友表示赞同！

罪歌

学习技术总是要从实际项目入手，这篇文章正好能展示结合这些技术的具体方法。

    有10位网友表示赞同！

微信名字

对图书查询检索服务感兴趣，想看看这个方案能不能帮助我解决类似的问题。

    有10位网友表示赞同！

强辩

Elasticsearch的搜索能力一直很强大，配合Spring Boot开发更方便。

    有7位网友表示赞同！

几妆痕

Logstash在数据采集方面做得很好，这篇文章提到的实现方式应该很有借鉴意义。

    有13位网友表示赞同！

掉眼泪

学习完这篇文章后是不是能自己搭建一个类似的图书查询系统呢?

    有8位网友表示赞同！

经典的对白

感觉这个技术栈很适合做海量数据的搜索和分析。

    有14位网友表示赞同！

该用户已上天

分享一下你使用Elasticsearch的时候遇到的问题和解决方案吗？

    有15位网友表示赞同！

回忆未来

最近在学习SpringBoot，看到这篇文章很有启发！

    有16位网友表示赞同！

蹂躏少女

对全文检索和数据采集比较感兴趣，这篇文章正好能增长见识。

    有16位网友表示赞同！

久爱不厌

希望能有更多实践案例分享，更直观地了解整套技术的实际运用。

    有20位网友表示赞同！

暖瞳

使用Spring Boot开发的服务速度应该很快吧？

    有14位网友表示赞同！

万象皆为过客

感觉这个技术方案对于大型图书馆的数字化管理非常有帮助。

    有18位网友表示赞同！

花容月貌

作者的代码示例讲解能详细点吗？ 希望能更好地理解思路和步骤 。

    有12位网友表示赞同！

抓不住i

这篇文章的内容看起来很系统化，从设计到实现都有详细介绍。

    有16位网友表示赞同！

苏莫晨

希望以后还能看到关于这个技术栈的更深入的探讨和应用案例。

    有19位网友表示赞同！

爱到伤肺i

想了解一下Elasticsearch的不同查询方式，以及如何根据需求进行优化？

    有17位网友表示赞同！

窒息

这篇文章的内容对学习Java架构很有帮助，感谢作者的不辞辛劳！

    有9位网友表示赞同！