前言

之前有段时间在研究MySQL的数据如何与Elasticsearch的数据之间如何进行表同步，后面经过我多番资料查找与学习，总算做出来了。本篇博客属于个人业余爱好，望大佬们不喜勿喷。

一、Elasticsearch是什么？

Elasticsearch是一个开源的高扩展的分布式全文检索引擎，它可以近乎实时的存储、检索数据；本身扩展性很好，可以扩展到上百台服务器，处理PB级别的数据。
Elasticsearch也使用Java开发并使用Lucene作为其核心来实现所有索引和搜索的功能，但是它的目的是通过简单的RESTful API来隐藏Lucene的复杂性，从而让全文搜索变得简单。

二、logstash

Logstash 是开源的服务器端数据处理管道，能够同时从多个来源采集数据，转换数据，然后将数据发送到 Elasticsearch。

1.下载地址

https://www.elastic.co/cn/downloads/logstash

在这里插入图片描述
下载对应的压缩包

2.解压之后的目录如下：

解压如下（示例）：

在这里插入图片描述

3.新建文件并配置导入数据源

1、在bin目录下新建.conf文件（名称随意），本博文以新建的logstash.conf文件为例：
在这里插入图片描述
2、修改logstash.conf（上文新建）配置文件，加入如下内容

input {
    stdin {
    }
    jdbc {
      # 配置数据库信息
      jdbc_connection_string => "jdbc:mysql://localhost:3306/exam?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai"
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_user => "root"
      jdbc_password => "111111"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      jdbc_default_timezone => "Asia/Shanghai"
  	   # mysql驱动所在位置
      jdbc_driver_library => "D:\mysql-connector-java\mysql-connector-java-5.0.4-bin.jar"
      #sql执行语句
	  statement => "SELECT * FROM exams"
      # 定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
      schedule => "* * * * *"
      # 是否将 sql 中 column 名称转小写
      lowercase_column_names => false
    }
}
output {
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "exam"
        # 文档_id，%{goods_id}意思是取查询出来的goods_id的值，并将其映射到shop的_id字段中
        # 文档_id，%{goodsId}如果是别名，意思是取查询出来的goodsId的值，并将其映射到shop的_id字段中
        document_id => "%{examId}"
    }
    stdout {
        codec => json_lines
    }
}

4.检查配置文件并启动logstash将数据导入elasticsearch

1、检查配置文件是否正确，在bin目录命令行窗口里输入：

logstash -f logstash.conf -t

在这里插入图片描述

Configuration OK 表示配置文件正确

2、启动logstash，在bin目录命令行窗口里输入：

logstash -f logstash.conf

启动成功之后，logstash会自动连接Mysql数据库，并将Mysql数据自动导入elasticsearch 中。

注意：导入数据前，先确保elasticsearch 已正常启动。

三、Springboot集成Elasticsearch

1.、导入Elasticsearch的依赖
在这里插入图片描述

2.、properties配置文件
在这里插入图片描述
3、Elasticsaerch对应的配置实体字段

4、Elasticsaerch的连接类

三、增加修改删除同步操作

1、增加操作可以使用上述logstash可以实现同步
2、修改操作可以使用上述logstash可以实现同步

3.删除操作同步

logstash尚不能实现删除同步操作，不能进行物理删除数据，我在这里提一个折中的方法，新增一个 flag列，标识记录是否已经被删除，这样，相同的记录也会存在于Elasticsearch。可以执行简单的term查询操作,检索出已经删除的数据信息。
其次，若需要执行clean up清理数据操作（物理删除），只需要在数据库和ES中同时删除掉标记位deleted的记录即可。如：mysql执行：delete from test where test.flag=‘deleted’; ES同样执行对应删除操作。