Elastic Search（ES）Java 入门实操（3）数据同步

2024-06-08 09:20
文章标签 java 数据 es 入门 实操 同步 search elastic

本文主要是介绍Elastic Search（ES）Java 入门实操（3）数据同步，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

基本概念和数据查询代码：

Elastic Search （ES）Java 入门实操（1）下载安装、概念-CSDN博客

Elastic Search（ES）Java 入门实操（2）搜索代码-CSDN博客

想要使用 ES 来查询数据，首先得要 ES 里有数据，但是如果是后来引入的 ES，数据库上万条的数据肯定不能通过手动进行同步，需要使用其他方法进行同步。

数据同步分为全量同步和增量同步。

所谓全量同步，就是引入 ES 时将 MySQL 里的数据全部同步到 ES 里。增量同步就是当数据库的数据发生变化时，将变化的数据同步到 ES 里。

同步方法

定时任务

通过定时任务的方式，每隔一段时间进行同步。比如每一分钟同步一次。

优点：简单，占用资源少，不用引入第三方中间件

缺点：有时间差，数据一致性要求高的场景不适用

全量同步通过实现 CommandLineRunner 接口，在程序启动时执行。

/*** CommandLineRunner 接口，当spring启动时就执行方法*/
@Component
public class FullSycnToEs implements CommandLineRunner {@Resourceprivate ArticleService articleService;@Resourceprivate ArticleEsDao articleEsDao;@Overridepublic void run(String... args) throws Exception {//spring 启动就执行方法进行全量同步//1.从MySQL获取数据List<Article> articleList = articleService.list();if(CollectionUtils.isEmpty(articleList)){return;}//2.将数据转换为DTOList<ArticleEsDto> articleDtoList = articleList.stream().map(ArticleEsDto::toDto).collect(Collectors.toList());//3.将数据同步到ESarticleEsDao.saveAll(articleDtoList);System.out.println("全量同步完成");}
}

增量同步使用 @ Scheduled 定时任务监控更新时间

注意启动类要加上注解 @EnableScheduling

/*** 定时任务执行数据同步*/
@Component
public class InSyncToEs {@Resourceprivate ArticleMapper articleMapper;@Resourceprivate ArticleEsDao articleEsDao;@Scheduled(fixedRate = 100)public void run(){// 定时任务，将数据同步到es，根据更新时间来判断//假定3分钟内，如果更新时间大于3分钟之前的时间，就是更新了，获取这个数据存入到ES 中Date minUpdateTime = new Date(new Date().getTime() - 5* 60*1000L);List<Article> newArticles = articleMapper.getNewArticles(minUpdateTime);//判断是否有数据更新if(CollectionUtils.isEmpty(newArticles)){//没有数据更新System.out.println("没有数据更新");return;}//有数据更新，将数据转换成dto格式List<ArticleEsDto> articleEsDtoList = newArticles.stream().map(ArticleEsDto::toDto).collect(Collectors.toList());//将数据存入到ES中articleEsDao.saveAll(articleEsDtoList);System.out.println("数据同步完成");}
}

双写

写入数据库时同时同步到 ES 中，需要考虑 ES 同步失败了怎么办。

使用事务来保证一致性，如果 ES 同步失败了，可以通过定时任务 + 日志 + 告警进行检测和修复（补偿）

Logstash 数据同步管道

传输和处理数据的管道

下载地址：https://artifacts.elastic.co/downloads/logstash/logstash-7.17.21-windows-x86_64.zip

官方文档：Jdbc input plugin | Logstash Reference [7.17] | Elastic

同样的，需要注意版本，下载解压之后在 config 文件夹创建新的同步文件，建议不同的同步脚本创建不同的文件，不要在同一个文件下配置。

文件配置根据官方文档修改，MySQL jar包使用绝对路径即可，否则可能找不到 jar 包，jar 包可以自行准备，也可以从项目的 maven 仓库获取。

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.input {jdbc {// MySQL jar包路径jdbc_driver_library => "D:\software\logstash-7.17.21\config\mysql-connector-java-8.0.29.jar"// MySQL 驱动jdbc_driver_class => "com.mysql.cj.jdbc.Driver"// MySQL 连接地址jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"//账号密码jdbc_user => "root"jdbc_password => "1234"//动态 SQLstatement => "SELECT * from article where 1=1"parameters => { "favorite_artist" => "Beethoven" }//定时执行，core 表达式schedule => "*/5 * * * * *"}
}output {stdout { codec => rubydebug }
}

 配置好之后在 logstash 目录下执行下面的命令，完成初步从数据库获取数据

.\bin\logstash.bat -f .\config\my-task.conf

 成功获取数据

增量同步配置，使用 updateTime 来进行同步更新的数据。

完整 input 配置如下。

input {jdbc {jdbc_driver_library => "D:\software\logstash-7.17.21\config\mysql-connector-java-8.0.29.jar"jdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"jdbc_user => "root"jdbc_password => "1234"// 动态查询语句，保证最后一条是最大的statement => "SELECT * from article where updateTime > :sql_last_value and updateTime < now() order by updateTime asc"// 查询参数的 hash，不用更改parameters => { "favorite_artist" => "Beethoven" }// 查询参数的类型，updatetime 是 timestamp 类型的tracking_column_type => "timestamp"// 查询参数tracking_column => "updatetime"// 设置为 true 时，将定义的查询参数值用作动态 SQL 中sql_last_value，false 时:sql_last_value 是上次查询时间use_column_value => true// 时区设置为上海，否则存在 8小时时差jdbc_default_timezone => "Asia/Shanghai"// core 表达式schedule => "*/5 * * * * *"}
}

配置好从 MySQL 获取的数据之后，就可以同步到 ES 中了。同样需要书写配置。

官方文档：Elasticsearch output plugin | Logstash Reference [7.17] | Elastic

output {stdout { codec => rubydebug }elasticsearch {//访问地址，就是本地 ES 端口hosts => "127.0.0.1:9200"// ES 索引index =>"article_1"// 数据 id，从数据库获取document_id => "%{id}"}

最终配置

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.input {jdbc {jdbc_driver_library => "D:\software\logstash-7.17.21\config\mysql-connector-java-8.0.29.jar"jdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"jdbc_user => "root"jdbc_password => "1234"statement => "SELECT * from article where updateTime > :sql_last_value and updateTime < now() order by updateTime asc "parameters => { "favorite_artist" => "Beethoven" }tracking_column_type => "timestamp"tracking_column => "updatetime"use_column_value => truejdbc_default_timezone => "Asia/Shanghai"schedule => "*/5 * * * * *"}
}
// 筛选
filter{mutate{//重命名rename => {"updatetime" =>"updateTime""createtime" => "createTime""isdetele" => "isDetele"}}
}output {stdout { codec => rubydebug }elasticsearch {hosts => "127.0.0.1:9200"index =>"article_1"document_id => "%{id}"}
}

同步成功！ 

logstash 的优点：配置完成后使用比较方便，插件多

                缺点：要多维护组件，一般需要配合其他中间件，比如（kafka）

Canal

下载地址：Releases · alibaba/canal (github.com)

文档：QuickStart · alibaba/canal Wiki (github.com)

实时同步数据，通过监控 MySQL 的 binlog，当数据库发生修改时，会修改 binlog 文件，然后 canal 监听到就可以同步到 ES 中。

对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，在 MySQL 目录下新建一个my.ini，配置如下

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant，直接在查询控制台执行如下命令

CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

bin 目录下 startup 启动即可。

然后 Java 需要一个客户端，首先引入依赖

<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.0</version>
</dependency>

客户端代码

import java.net.InetSocketAddress;
import java.util.List;import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;public class SimpleCanalClientExample {public static void main(String args[]) {// 创建链接CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),11111), "example", "", "");int batchSize = 1000;int emptyCount = 0;try {connector.connect();connector.subscribe(".*\\..*");connector.rollback();int totalEmptyCount = 120;while (emptyCount < totalEmptyCount) {Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据long batchId = message.getId();int size = message.getEntries().size();if (batchId == -1 || size == 0) {emptyCount++;System.out.println("empty count : " + emptyCount);try {Thread.sleep(1000);} catch (InterruptedException e) {}} else {emptyCount = 0;// System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);printEntry(message.getEntries());}connector.ack(batchId); // 提交确认// connector.rollback(batchId); // 处理失败, 回滚数据}System.out.println("empty too many times, exit");} finally {connector.disconnect();}
}private static void printEntry(List<Entry> entrys) {for (Entry entry : entrys) {if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {continue;}RowChange rowChage = null;try {rowChage = RowChange.parseFrom(entry.getStoreValue());} catch (Exception e) {throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),e);}EventType eventType = rowChage.getEventType();System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),eventType));for (RowData rowData : rowChage.getRowDatasList()) {if (eventType == EventType.DELETE) {printColumn(rowData.getBeforeColumnsList());} else if (eventType == EventType.INSERT) {printColumn(rowData.getAfterColumnsList());} else {System.out.println("-------&gt; before");printColumn(rowData.getBeforeColumnsList());System.out.println("-------&gt; after");printColumn(rowData.getAfterColumnsList());}}}
}private static void printColumn(List<Column> columns) {for (Column column : columns) {System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());}
}}

 

过程出现的问题

1. 在执行命令.\bin\logstash.bat -f .\config\my-task.conf  时报错

只需要更改 bin 目录下的 setup.bat 文件中的双引号去掉即可。 

2. canal 启动 报错

修改变量或者修改启动项

编辑 startup.bat，在文件中添加如下配置：

// 自己的 jdk 路径
set JAVA_HOME=C:\Users\p'b\.jdks\corretto-1.8.0_392
// 覆盖环境变量
set PATH=%JAVA_HOME%\bin;%PATH%

 

这篇关于Elastic Search（ES）Java 入门实操（3）数据同步的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1041777。

相关文章

Java中流式并行操作parallelStream的原理和使用方法

Java中流式并行操作parallelStream的原理和使用方法

《Java中流式并行操作parallelStream的原理和使用方法》本文详细介绍了Java中的并行流（parallelStream）的原理、正确使用方法以及在实际业务中的应用案例,并指出在使用并行流... 目录Java中流式并行操作parallelStream0. 问题的产生1. 什么是parallelS
阅读更多...
Java中Redisson 的原理深度解析

Java中Redisson 的原理深度解析

《Java中Redisson的原理深度解析》Redisson是一个高性能的Redis客户端,它通过将Redis数据结构映射为Java对象和分布式对象,实现了在Java应用中方便地使用Redis,本文... 目录前言一、核心设计理念二、核心架构与通信层1. 基于 Netty 的异步非阻塞通信2. 编解码器三、
阅读更多...
SpringBoot基于注解实现数据库字段回填的完整方案

SpringBoot基于注解实现数据库字段回填的完整方案

《SpringBoot基于注解实现数据库字段回填的完整方案》这篇文章主要为大家详细介绍了SpringBoot如何基于注解实现数据库字段回填的相关方法,文中的示例代码讲解详细,感兴趣的小伙伴可以了解... 目录数据库表pom.XMLRelationFieldRelationFieldMapping基础的一些代
阅读更多...
一篇文章彻底搞懂macOS如何决定java环境

一篇文章彻底搞懂macOS如何决定java环境

《一篇文章彻底搞懂macOS如何决定java环境》MacOS作为一个功能强大的操作系统,为开发者提供了丰富的开发工具和框架,下面：本文主要介绍macOS如何决定java环境的相关资料,文中通过代码... 目录方法一：使用 which命令方法二：使用 Java_home工具（Apple 官方推荐）那问题来了，
阅读更多...
Java HashMap的底层实现原理深度解析

Java HashMap的底层实现原理深度解析

《JavaHashMap的底层实现原理深度解析》HashMap基于数组+链表+红黑树结构,通过哈希算法和扩容机制优化性能,负载因子与树化阈值平衡效率,是Java开发必备的高效数据结构,本文给大家介绍... 目录一、概述：HashMap的宏观结构二、核心数据结构解析1. 数组（桶数组）2. 链表节点（Node
阅读更多...
Java AOP面向切面编程的概念和实现方式

Java AOP面向切面编程的概念和实现方式

《JavaAOP面向切面编程的概念和实现方式》AOP是面向切面编程,通过动态代理将横切关注点（如日志、事务）与核心业务逻辑分离,提升代码复用性和可维护性,本文给大家介绍JavaAOP面向切面编程的概... 目录一、AOP 是什么？二、AOP 的核心概念与实现方式核心概念实现方式三、Spring AOP 的关
阅读更多...
详解SpringBoot+Ehcache使用示例

详解SpringBoot+Ehcache使用示例

《详解SpringBoot+Ehcache使用示例》本文介绍了SpringBoot中配置Ehcache、自定义get/set方式,并实际使用缓存的过程,文中通过示例代码介绍的非常详细,对大家的学习或者... 目录摘要概念内存与磁盘持久化存储：配置灵活性：编码示例引入依赖：配置ehcache.XML文件：配置
阅读更多...
Java 虚拟线程的创建与使用深度解析

Java 虚拟线程的创建与使用深度解析

《Java虚拟线程的创建与使用深度解析》虚拟线程是Java19中以预览特性形式引入,Java21起正式发布的轻量级线程,本文给大家介绍Java虚拟线程的创建与使用,感兴趣的朋友一起看看吧... 目录一、虚拟线程简介1.1 什么是虚拟线程？1.2 为什么需要虚拟线程？二、虚拟线程与平台线程对比代码对比示例：三
阅读更多...
Java中的.close()举例详解

Java中的.close()举例详解

《Java中的.close()举例详解》.close()方法只适用于通过window.open()打开的弹出窗口,对于浏览器的主窗口,如果没有得到用户允许是不能关闭的,：本文主要介绍Java中的.... 目录当你遇到以下三种情况时，一定要记得使用 .close()：用法作用举例如何判断代码中的 input
阅读更多...
Linux下利用select实现串口数据读取过程

Linux下利用select实现串口数据读取过程

《Linux下利用select实现串口数据读取过程》文章介绍Linux中使用select、poll或epoll实现串口数据读取,通过I/O多路复用机制在数据到达时触发读取,避免持续轮询,示例代码展示设... 目录示例代码（使用select实现）代码解释总结在 linux 系统里，我们可以借助 select、
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2