MySQL如何实时同步数据到ES？试试这款阿里开源的神器 SpringBoot实战电商项目mall（40

SpringBoot实战电商项目mall（40k+star）地址：github.com/macrozheng/…
摘要mall项目中的商品搜索功能，一直都没有做实时数据同步。最近发现阿里巴巴开源的canal可以把MySQL中的数据实时同步到Elasticsearch中，能很好地解决数据同步问题。今天我们来讲讲canal的使用，希望对大家有所帮助！
canal简介canal主要用途是对MySQL数据库增量日志进行解析，提供增量数据的订阅和消费，简单说就是可以对MySQL的增量数据进行实时同步，支持同步到MySQL、Elasticsearch、HBase等数据存储中去。
canal工作原理canal会模拟MySQL主库和从库的交互协议，从而伪装成MySQL的从库，然后向MySQL主库发送dump协议， MySQL主库收到dump请求会向canal推送binlog ， canal通过解析binlog将数据同步到其他存储中去。
文章插图
canal使用接下来我们来学习下canal的使用，以MySQL实时同步数据到Elasticsearch为例。
组件下载

首先我们需要下载canal的各个组件canal-server、canal-adapter、canal-admin ，下载地址：

文章插图
【MySQL如何实时同步数据到ES？试试这款阿里开源的神器】canal的各个组件的用途各不相同，下面分别介绍下：

canal-server（canal-deploy）：可以直接监听MySQL的binlog ，把自己伪装成MySQL的从库，只负责接收数据，并不做处理。
canal-adapter：相当于canal的客户端，会从canal-server中获取数据，然后对数据进行同步，可以同步到MySQL、Elasticsearch和HBase等存储中去。
canal-admin：为canal提供整体配置管理、节点运维等面向运维的功能，提供相对友好的WebUI操作界面，方便更多用户快速和安全的操作。

由于不同版本的MySQL、Elasticsearch和canal会有兼容性问题，所以我们先对其使用版本做个约定。

文章插图
MySQL配置

由于canal是通过订阅MySQL的binlog来实现数据同步的，所以我们需要开启MySQL的binlog写入功能，并设置binlog-format为ROW模式，我的配置文件为/mydata/mysql/conf/my.cnf ，改为如下内容即可；

[mysqld]## 设置server_id ， 同一局域网中需要唯一server_id=101 ## 指定不需要同步的数据库名称binlog-ignore-db=mysql## 开启二进制日志功能log-bin=mall-mysql-bin## 设置二进制日志使用内存大小（事务）binlog_cache_size=1M## 设置使用的二进制日志格式（mixed,statement,row）binlog_format=row## 二进制日志过期清理时间 。 默认值为0 ， 表示不自动清理 。 expire_logs_days=7## 跳过主从复制中遇到的所有错误或指定类型的错误 ， 避免slave端复制中断 。 ## 如：1062错误是指一些主键重复 ， 1032错误是因为主从数据库数据不一致slave_skip_errors=1062

配置完成后需要重新启动MySQL ，重启成功后通过如下命令查看binlog是否启用；

show variables like '%log_bin%'

+---------------------------------+-------------------------------------+| Variable_name| Value|+---------------------------------+-------------------------------------+| log_bin| ON|| log_bin_basename| /var/lib/mysql/mall-mysql-bin|| log_bin_index| /var/lib/mysql/mall-mysql-bin.index || log_bin_trust_function_creators | OFF|| log_bin_use_v1_row_events| OFF|| sql_log_bin| ON|+---------------------------------+-------------------------------------+

再查看下MySQL的binlog模式；

show variables like 'binlog_format%';

+---------------+-------+| Variable_name | Value |+---------------+-------+| binlog_format | ROW|+---------------+-------+

接下来需要创建一个拥有从库权限的账号，用于订阅binlog ，这里创建的账号为canal:canal；

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

创建好测试用的数据库canal-test ，之后创建一张商品表product ，建表语句如下。

CREATE TABLE `product`(`id` bigint(20) NOT NULL AUTO_INCREMENT,`title` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,`sub_title` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,`price` decimal(10, 2) NULL DEFAULT NULL,`pic` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,PRIMARY KEY (`id`) USING BTREE) ENGINE = InnoDB AUTO_INCREMENT = 2 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

canal-server使用