使用canal增量同步ES索引库数据

2024-08-24 02:44
文章标签 数据 es 使用 索引 canal 同步 增量

本文主要是介绍使用canal增量同步ES索引库数据，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Canal增量数据同步利器

Canal介绍

canal主要用途是基于 MySQL 数据库增量日志解析，并能提供增量数据订阅和消费，应用场景十分丰富。

github地址：https://github.com/alibaba/canal

版本下载地址：https://github.com/alibaba/canal/releases

文档地址：https://github.com/alibaba/canal/wiki/Docker-QuickStart

Canal应用场景

1.电商场景下商品、用户实时更新同步到至Elasticsearch、solr等搜索引擎；
2.价格、库存发生变更实时同步到redis；
3.数据库异地备份、数据同步；
4.代替使用轮询数据库方式来监控数据库变更，有效改善轮询耗费数据库资源。
在这里插入图片描述

MySQL主从复制原理

1.MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
2.MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
3.MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

Canal工作原理

1.canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
2.MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
3.canal 解析 binary log 对象(原始为 byte 流)
在这里插入图片描述

Canal安装

参考文档：https://github.com/alibaba/canal/wiki/QuickStart

MySQL Bin-log开启

1)MySQL开启bin-log

a.进入mysql容器

docker exec -it -u root mysql /bin/bash

b.开启mysql的binlog

cd /etc/mysql/mysql.conf.d在mysqld.cnf最下面添加如下配置
# 开启 binlog
log-bin=/var/lib/mysql/mysql-bin
# 选择 ROW 模式
binlog-format=ROW
# 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复
server-id=12345

c.创建账号并授权

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant：

create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

d.重启mysql

docker restart mysql

开启bin-log后，我们可以用sql语句查看下：

show variables like '%log_bin%'

效果如下：
在这里插入图片描述

Canal安装

1)拉取镜像

docker pull canal/canal-server:v1.1.1

2)安装容器

a.安装canal-server容器

docker run -p 11111:11111 --name canal -d docker.io/canal/canal-server

b.配置canal-server

修改/home/admin/canal-server/conf/canal.properties,将它的id属性修改成和mysql数据库中server-id不同的值，如下图：
在这里插入图片描述
c.修改/home/admin/canal-server/conf/example/instance.properties,配置要监听的数据库服务地址和监听数据变化的数据库以及表，修改如下：
在这里插入图片描述
在这里插入图片描述
指定监听数据库表的配置如下canal.instance.filter.regex：

mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\) 
常见例子：
1.  所有表：.*   or  .*\\..*
2.  canal schema下所有表： canal\\..*
3.  canal下的以canal打头的表：canal\\.canal.*
4.  canal schema下的一张表：canal.test1
5.  多个规则组合使用：canal\\..*,mysql.test1,mysql.test2 (逗号分隔)
注意：此过滤条件只针对row模式的数据有效(ps. mixed/statement因为不解析sql，所以无法准确提取tableName进行过滤)

重启canal:

docker restart canal
Canal微服务

​ 我们搭建一个微服务，用于读取canal监听到的变更日志，微服务名字叫seckill-canal。该项目我们需要引入canal-spring-boot-autoconfigure包，并且需要实现EntryHandler接口，该接口中有3个方法，分别为insert、update、delete,这三个方法用于监听数据增删改变化。

参考地址：https://github.com/NormanGyllenhaal/canal-client

1)pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<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"><parent><artifactId>seckill-service</artifactId><groupId>com.seckill</groupId><version>0.0.1-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>seckill-canal</artifactId><dependencies><!--web--><dependency><groupId>com.seckill</groupId><artifactId>seckill-web</artifactId><version>0.0.1-SNAPSHOT</version></dependency><!--esAPI--><dependency><groupId>com.seckill</groupId><artifactId>seckill-search-api</artifactId><version>0.0.1-SNAPSHOT</version></dependency><!--goodsAPI--><dependency><groupId>com.seckill</groupId><artifactId>seckill-goods-api</artifactId><version>0.0.1-SNAPSHOT</version></dependency><!--canal--><dependency><groupId>top.javatool</groupId><artifactId>canal-spring-boot-autoconfigure</artifactId><version>1.2.1-RELEASE</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><!-- 指定该Main Class为全局的唯一入口 --><mainClass>com.seckill.CanalApplication</mainClass><layout>ZIP</layout></configuration><executions><execution><goals><goal>repackage</goal><!--可以把依赖的包都打包到生成的Jar包中--></goals></execution></executions></plugin></plugins></build>
</project>

bootstrap.yml配置

server:port: 18088
spring:application:name: seckill-canalcloud:nacos:config:file-extension: yamlserver-addr: nacos-server:8848discovery:#Nacos的注册地址server-addr: nacos-server:8848
#超时配置
ribbon:ReadTimeout: 3000000
#Canal配置
canal:server: canal-server:11111destination: example
#日志
logging:level:root: error

2)创建com.seckill.handler.SkuHandler实现EntryHandler接口，代码如下：

@Component
@CanalTable(value = "tb_sku")
public class SkuHandler implements EntryHandler<Sku> {/**** 增加数据* @param sku*/@Overridepublic void insert(Sku sku) {System.out.println("===========insert:"+sku);}/**** 修改数据* @param before* @param after*/@Overridepublic void update(Sku before, Sku after) {System.out.println("===========update-before:"+before);System.out.println("===========update-after:"+after);}/**** 删除数据* @param sku*/@Overridepublic void delete(Sku sku) {System.out.println("===========delete:"+sku);}
}

3)创建启动类

@SpringBootApplication
public class CanalApplication {public static void main(String[] args) {SpringApplication.run(CanalApplication.class,args);}
}

程序启动后，修改tb_sku数据，可以看到控制会打印修改前后的数据：
在这里插入图片描述

索引库同步

当tb_sku秒杀商品发生变化时，我们应该同时变更索引库中的索引数据，比如秒杀商品增加，则需要同步增加秒杀商品的索引，如果有秒杀商品删除，则需要同步移除秒杀商品。

修改seckill-canal中的com.seckill.handler.SkuHandler的增删改方法，代码如下：

@Component
@CanalTable(value = "tb_sku")
public class SkuHandler implements EntryHandler<Sku> {@Autowiredprivate SkuInfoFeign skuInfoFeign;/**** 增加数据* @param sku*/@Overridepublic void insert(Sku sku) {//将Sku转换成SkuInfoSkuInfo skuInfo = JSON.parseObject( JSON.toJSONString(sku) ,SkuInfo.class);//同步索引skuInfoFeign.modify(1,skuInfo);}/**** 修改数据* @param before* @param after*/@Overridepublic void update(Sku before, Sku after) {int type=2;//将Sku转换成SkuInfoSkuInfo skuInfo = JSON.parseObject( JSON.toJSONString(after) ,SkuInfo.class);if(skuInfo.getStatus()==1 || after.getSeckillNum()<=0){//商品变成了普通商品,或者商品库存为0，则需要删除索引数据type=3;}//同步索引skuInfoFeign.modify(type,skuInfo);}/**** 删除数据* @param sku*/@Overridepublic void delete(Sku sku) {//将Sku转换成SkuInfoSkuInfo skuInfo = JSON.parseObject( JSON.toJSONString(sku) ,SkuInfo.class);//同步索引skuInfoFeign.modify(3,skuInfo);}
}

开启Feign功能：@EnableFeignClients(basePackages = {“com.seckill.search.feign”})
在这里插入图片描述
此时对数据库中tb_sku表进行增删改的时候，会同步到索引库中。

这篇关于使用canal增量同步ES索引库数据的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

解决pandas无法读取csv文件数据的问题

解决pandas无法读取csv文件数据的问题

《解决pandas无法读取csv文件数据的问题》本文讲述作者用Pandas读取CSV文件时因参数设置不当导致数据错位,通过调整delimiter和on_bad_lines参数最终解决问题,并强调正确参... 目录一、前言二、问题复现1. 问题2. 通过 on_bad_lines=‘warn’ 跳过异常数据3
阅读更多...
Android Paging 分页加载库使用实践

Android Paging 分页加载库使用实践

《AndroidPaging分页加载库使用实践》AndroidPaging库是Jetpack组件的一部分,它提供了一套完整的解决方案来处理大型数据集的分页加载,本文将深入探讨Paging库... 目录前言一、Paging 库概述二、Paging 3 核心组件1. PagingSource2. Pager3.
阅读更多...
python使用try函数详解

python使用try函数详解

《python使用try函数详解》Pythontry语句用于异常处理,支持捕获特定/多种异常、else/final子句确保资源释放,结合with语句自动清理,可自定义异常及嵌套结构,灵活应对错误场景... 目录try 函数的基本语法捕获特定异常捕获多个异常使用 else 子句使用 finally 子句捕获所
阅读更多...
C++11右值引用与Lambda表达式的使用

C++11右值引用与Lambda表达式的使用

《C++11右值引用与Lambda表达式的使用》C++11引入右值引用,实现移动语义提升性能,支持资源转移与完美转发；同时引入Lambda表达式,简化匿名函数定义,通过捕获列表和参数列表灵活处理变量... 目录C++11新特性右值引用和移动语义左值 / 右值常见的左值和右值移动语义移动构造函数移动复制运算符
阅读更多...
Python对接支付宝支付之使用AliPay实现的详细操作指南

Python对接支付宝支付之使用AliPay实现的详细操作指南

《Python对接支付宝支付之使用AliPay实现的详细操作指南》支付宝没有提供PythonSDK,但是强大的github就有提供python-alipay-sdk,封装里很多复杂操作,使用这个我们就... 目录一、引言二、准备工作2.1 支付宝开放平台入驻与应用创建2.2 密钥生成与配置2.3 安装ali
阅读更多...
C#中lock关键字的使用小结

C#中lock关键字的使用小结

《C#中lock关键字的使用小结》在C#中,lock关键字用于确保当一个线程位于给定实例的代码块中时,其他线程无法访问同一实例的该代码块,下面就来介绍一下lock关键字的使用... 目录使用方式工作原理注意事项示例代码为什么不能lock值类型在C#中，lock关键字用于确保当一个线程位于给定实例的代码块中时
阅读更多...
MySQL 强制使用特定索引的操作

MySQL 强制使用特定索引的操作

《MySQL强制使用特定索引的操作》MySQL可通过FORCEINDEX、USEINDEX等语法强制查询使用特定索引,但优化器可能不采纳,需结合EXPLAIN分析执行计划,避免性能下降,注意版本差异... 目录1. 使用FORCE INDEX语法2. 使用USE INDEX语法3. 使用IGNORE IND
阅读更多...
C# $字符串插值的使用

C# $字符串插值的使用

《C#$字符串插值的使用》本文介绍了C#中的字符串插值功能,详细介绍了使用$符号的实现方式,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习吧... 目录$ 字符使用方式创建内插字符串包含不同的数据类型控制内插表达式的格式控制内插表达式的对齐方式内插表达式中使用转义序列内插表达式中使用
阅读更多...
flask库中sessions.py的使用小结

flask库中sessions.py的使用小结

《flask库中sessions.py的使用小结》在Flask中Session是一种用于在不同请求之间存储用户数据的机制,Session默认是基于客户端Cookie的,但数据会经过加密签名,防止篡改,... 目录1. Flask Session 的基本使用(1) 启用 Session(2) 存储和读取 Se
阅读更多...
Java Thread中join方法使用举例详解

Java Thread中join方法使用举例详解

《JavaThread中join方法使用举例详解》JavaThread中join()方法主要是让调用改方法的thread完成run方法里面的东西后,在执行join()方法后面的代码,这篇文章主要介绍... 目录前言1.join()方法的定义和作用2.join()方法的三个重载版本3.join()方法的工作原
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2