当前位置：首页 > news >正文

使用 Canal 实现 MySQL 与 ES 数据同步的技术实践

news 来源：原创 2025/9/21 7:02:37

前言

本文将详细讲解如何使用阿里的 Canal 工具，实现 MySQL 向 ES（Elasticsearch）的数据同步。

数据同步有多种方式，双写同步数据方式因性能慢、存在分布式事务及数据一致性问题、业务耦合度高且难以扩展，不适合采用。而异步同步数据方案中，如阿里的 Canal 和 Debezium 工具等，它们基于 CDC（数据抓取变更），监听 binlog 日志实现同步。相比之下，Debezium 需要集成 Kafka 且需手写消费者代码，系统复杂度较高，因此本文选择使用 Canal 进行数据同步。

一、基础知识

1.1 主从复制原理

MySQL 的主从复制依赖于 binlog，它记录 MySQL 上的所有变化，并以二进制形式保存在磁盘上的二进制日志文件中。主从复制过程是将 binlog 中的数据从主库传输到从库，一般为异步过程。

详细流程如下：

主库写 binlog：主库的更新 SQL（update、insert、delete）被写入 binlog。
主库发送 binlog：主库创建一个 log dump 线程，用于发送 binlog 给从库。
从库写 relay log：从库连接到主节点时，创建一个 IO 线程，请求主库更新的 binlog，并将接收到的 binlog 信息写入 relay log 日志文件。
从库回放：从库创建一个 SQL 线程，读取 relay log 中的内容，并在从库中回放，实现主从数据一致性。

1.2 Canal 基础

Canal 是一款常用的数据同步工具，基于 Binlog 订阅方式实现，模拟 MySQL Slave 订阅 Binlog 日志，实现 CDC（Change Data Capture），将已提交的更改发送到下游。

主要流程如下：

Canal 服务端向 MySQL 的 master 节点传输 dump 协议。
MySQL 的 master 节点接收到 dump 请求后，推送 Binlog 日志给 Canal 服务端，并将 Binlog 对象（原始为 byte 流）解析转成 Json 格式。
Canal 客户端通过 TCP 协议或 MQ 形式监听 Canal 服务端，将数据同步到 ES。

Canal 执行的核心流程中，Binlog Parser 负责 Binlog 的提取、解析和推送，EventSink 负责数据的过滤、路由和加工。

二、软件下载安装

2.1 Java JDK

官网：[https://www.oracle.com/java/technologies/downloads/]
推荐版本：11.0.19 。由于Canal和ES的安装都强依赖JDK，需注意版本选择，不合适的版本可能导致ES安装失败以及Canal同步数据到ES时出现问题。
安装步骤：
1. 下载JDK安装包后，双击运行安装程序。
2. 在安装向导中，按照提示选择安装路径等配置，一般可使用默认路径，也可自定义合适路径。
3. 安装完成后，右键点击“此电脑”，选择“属性”，点击“高级系统设置”，在弹出的“系统属性”窗口中，选择“高级”选项卡，点击“环境变量”。
4. 在“系统变量”中找到“Path”变量，点击“编辑”，新建变量值，将JDK安装目录下的“bin”文件夹路径添加进去（例如：C:\Program Files\Java\jdk - 11.0.19\bin ）。
5. 新建“JAVA_HOME”系统变量，变量值为JDK的安装目录（例如：C:\Program Files\Java\jdk - 11.0.19 ）。
6. 打开命令提示符（CMD），输入“java -version”，若能正确显示JDK版本信息，说明安装配置成功。

2.2 MySQL

下载：前往MySQL官方网站（https://dev.mysql.com/downloads/mysql/ ）下载Windows版本的MySQL安装包，可根据系统是32位还是64位选择合适版本。
安装步骤：
1. 运行安装程序，选择“Custom”（自定义）安装类型，点击“Next”。
2. 在“Product Configuration”中，选择要安装的MySQL组件，一般保持默认勾选的“MySQL Server”等必要组件即可，点击“Next”。
3. 选择安装路径，可使用默认路径，也可自定义，点击“Next”。
4. 进入“Type and Networking”配置界面，一般保持默认的“Server Type”（服务器类型）和端口设置（默认3306），点击“Next”。
5. 在“Authentication Method”（认证方式）中，可选择“Use Legacy Authentication Method (Retain MySQL 5.x Compatibility)”（使用旧版认证方式，保持与MySQL 5.x兼容），设置好root用户密码等信息后点击“Next”。
6. 确认安装信息无误后，点击“Execute”开始安装，安装完成后点击“Finish”。
开启BinLog：
1. 找到MySQL安装目录下的“my.ini”（如果没有则新建一个）文件，使用文本编辑器打开。
2. 在文件中添加或修改以下配置：

[mysqld]
log - bin = mysql - bin
server - id = 1

3. 保存文件后，打开“服务”（可通过在“运行”中输入“services.msc”打开 ），找到“MySQL”服务，右键点击选择“重启”。
4. 重启后，可通过登录MySQL命令行（在命令提示符中输入“mysql - uroot - p”，输入密码登录 ），执行“show variables like '%log_bin%';”命令，若“Value”为“ON”，说明BinLog已开启成功。

创建Canal使用账号：登录MySQL命令行，执行以下SQL语句创建账号并赋予权限：

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'localhost' IDENTIFIED BY 'canal';
FLUSH PRIVILEGES;

2.3 Canal

官网：[https://github.com/alibaba/canal/releases]
版本：v1.1.6 ，需下载canal.adapter和canal.deployer两个部分。
- canal.deployer：相当于canal的服务端，启动它才可以在客户端接收数据库变更信息。
- canal.adapter：增加客户端数据落地的适配及启动功能（当deployer接收到消息后，会根据不同的目标源做适配，比如是es目标源适配和hbase适配等等）。
安装步骤：
1. 下载后，将canal.deployer和canal.adapter解压到合适的目录（建议目录路径不要包含中文和特殊字符）。
2. 对于canal.deployer，进入其解压后的目录，后续配置修改和启动操作在此目录进行。
3. 对于canal.adapter，同样进入其解压后的目录，后续配置修改和启动操作在此目录进行。

2.4 ES

官网：[https://www.elastic.co/cn/downloads/elasticsearch]
版本：7.17.4
安装步骤：
1. 下载Windows版本的ES安装包（一般为zip格式），解压到合适的目录（建议目录路径不要包含中文和特殊字符）。
2. 进入解压后的“bin”文件夹，双击“elasticsearch.bat”启动ES。
3. 启动成功后，打开浏览器，输入“http://localhost:9200/?pretty”，若能看到ES的相关信息返回，说明ES启动成功。

2.5 Kibana

下载网址：[https://www.elastic.co/cn/downloads/past-releases#kibana]
版本：7.14.0
安装步骤：
1. 下载Windows版本的Kibana安装包（一般为zip格式），解压到合适的目录（建议目录路径不要包含中文和特殊字符）。
2. 进入解压后的“bin”文件夹，双击“kibana.bat”启动Kibana。
3. 启动成功后，打开浏览器，输入“http://localhost:5601/app/dev_tools#/console”，若能进入Kibana操作界面，说明Kibana启动成功。

2.6 IK分词器

下载网址：[https://github.com/medcl/elasticsearch-analysis-ik/releases/tag/v7.17.2]
版本：v7.17.2
安装步骤：
1. 下载与ES版本对应的IK分词器压缩包，解压得到相关文件。
2. 在ES安装目录下找到“plugins”文件夹，在其下新建“analysis - ik”文件夹（如果没有）。
3. 将解压后的IK分词器文件复制到“analysis - ik”文件夹中。
4. 重启ES，使IK分词器生效。可通过在Kibana的Dev Tools中执行相关DSL语句测试分词效果，例如：

POST /_analyze
{"analyzer": "ik_max_word","text": "这是一段测试文本"
}

三、Canal 配置

3.1 canal.deployer 配置

修改 conf—>example 文件夹中的 instance.properties 文件，配置监听的数据库信息，主要修改监听 MySQL 的 URL、用户名和密码（默认账号密码为 canal）。

3.2 canal.deployer 启动

在 canal.deployer 中的 bin 文件目录下，执行启动命令 startup.sh（Windows环境执行startup.bat脚本）。启动成功后，可查看启动日志，“start successful”表示已成功启动，并开始监听 MySQL 数据库。

3.3 canal.adapte 配置

Step1：将 adapter 下面的 bootstrap.yml 文件内容全部注释掉，否则会提示表不存在。

Step2：修改 adapter 的 application.yml 配置文件。需注意 mysql 的账号密码是否正确，以及 es 链接是否添加“http://”前缀。

Step3：修改在 application.yml 中配置的目标数据源 es7 文件夹内容。以监听表 article 为例，在 es7 文件夹中创建 article.yml 文件，配置如下：

dataSourceKey :defaultDS #源数据源的key , 对应上面配置的srcDataSources中的值
destination :example #canal的instance或者MQ的topic
groupId :g1 #对应MQ模式下的groupId , 只会同步对应groupId的数据
esMapping :_index :article #es 的索引名称_id :_id #es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配sql : "SELECT t .idAS _id , t .id, t .user_id, t .article_type , t .title, t .short_title, t .picture,t .summary , t .category_id, t .source, t .source_url, t .offical_stat , t .topping_stat,t .cream_stat, t .`status`, t .deleted, t .create_time, t .update_timeFROM article t "        #sql映射commitBatch :1 #提交批大小

Step4：在 Kibana 中创建 ES 的 article 索引，代码如下：

PUT /article
{"mappings" : {"properties" : {"id" : {"type" : "integer"},"user_id" : {"type" : "integer"},"article_type" : {"type" : "integer"},"title" : {"type" : "text","analyzer": "ik_max_word"},"short_title" : {"type" : "text","analyzer": "ik_max_word"},"picture" : {"type" : "text","analyzer": "ik_max_word"},"summary" : {"type" : "text","analyzer": "ik_max_word"},"category_id" : {"type" : "integer"},"source" : {"type" : "integer"},"source_url" : {}}}
}

3.4 canal.adapte 启动

启动后查看启动日志，若没有报错且启动了 8081 端口，则表示启动成功，此时可进行操作。

四、数据同步实战

4.1 全量同步

启动 adapter 后，可使用源码中提供的接口进行全量数据同步，执行同步命令后，会提示导入的数据条数。

curl http://127.0.0.1:8081/etl/es7/article.yml -X POST

4.2 增量同步

当在 MySQL 中进行 update、delete 和 insert 操作时，ES 中的数据会相应变化。例如，进行修改操作后，查看 ES 查询结果，可验证数据已成功更改。

五、总结

通过本文，我们详细了解了使用 Canal 实现 MySQL 到 ES 数据同步的全过程，包括基础知识、软件安装、Canal 配置以及全量和增量数据同步实战。希望这些内容能为大家在实际项目中应用该技术提供帮助。如果项目中遇到类似场景，基本可以直接参考本文的方法进行操作。

以上就是本次技术分享的全部内容，希望对大家有所帮助！

前言