Canal实现Mysql和ES数据同步

向右看齐 2023-10-10 15:21 30阅读 0赞

1.ES概述

ES全称为"ElasticSewrch", 是一个基于RESTful web接口, 并且构建在Apache Lucene之上的开源分布式搜索引擎。

ES可以在极短的时间内存储、搜索和分析大量的数据。通常作为具有复杂搜索场景情况下的核心发动机。

简单来说, ES可以帮助我们快速完成海量数据的搜索工作。

2.场景介绍

在实际开发中,为了保证数据的安全性和持久性, 所以,我们一般会把数据存储的数据库中, 比如mysql数据库。

但, 如果数据量比较大, 并且搜索业务比较多, 则需要使用ES来实现我们的搜索功能。

而ES所实现的搜索, 需要把数据放入ES中, 才能实现, 所以, 就需要我们把mysql中的数据, 同步到ES中才可以。

3.实现方案

实现Mysql与ES之间的数据同步, 大体上可以分为三种方案:

方案一：同步调用

![bb4b257695193c54d2d9de71a1b1b3f7.jpeg][]

方案二：异步通知

![8dcd25f9a245ec8aa94d1c3fd3d194d5.jpeg][]

方案三：监听binlog

![de0efdacd6561e8ac4bc4c7300f43964.jpeg][]

4.监听binlog

在以上三种实现方案中, 可靠性相对来说比较好的是第三种方案"监听binlog",

| 方案 | 优势 | 缺点 |

| ---------- | ------------------------------ | -------------------------------------- |

| 同步调用 | 实现简单，粗暴 | 业务耦合度高 |

| 异步通知 | 低耦合，实现难度一般 | 依赖mq的可靠性 |

| 监听binlog | 完全解除服务间耦合, 可靠性较强 | 开启binlog增加数据库负担、实现复杂度高 |

4.1 实现原理

binlog其实就是Binary Log, 是MySQL二进制日志, 也可叫作变更日志(Update Log), 是 MySQL 中非常重要的日志。

主要用于记录数据库的变化情况，即 SQL语句的DDL和DML语句，不包含数据记录查询操作。

BinaryLog通常用于数据库的主从复制. 其工作原理如下:

![2f0036f88bcf5e9a93953072e9ddbb0e.jpeg][]

- 1)MySQL master 将数据变更写入二进制日志( binary log)，其中记录的数据叫做binary log events

- 2)MySQL slave 将 master 的 binary log events拷贝到它的中继日志(relay log)

- 3)MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

4.2 实现方案-Canal

Canal是阿里巴巴旗下的一款开源项目, 基于Java开发。Canal就是监听数据库的binary log，从而提供增量数据订阅&消费。

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

Canal就是把自己伪装成MySQL的一个slave节点，从而监听master的binary log变化。

Canal会把得到的变化信息通知给Canal的客户端，进而完成对其它数据库的同步。

5.搭建环境

5.1 开启MySQL主从

① 修改配置文件

找到mysql的配置文件my.cnf,并在上边添加配置信息

log-bin=/var/lib/mysql/mysql-bin
    binlog-do-db=heima

配置解读：

- \`log-bin=/var/lib/mysql/mysql-bin\`：设置binary log文件的存放地址和文件名，叫做mysql-bin

- \`binlog-do-db=heima\`：指定对哪个database记录binary log events，这里记录heima这个库

以Docker容器的Mysql为例,最终配置：

[mysqld]
    skip-name-resolve
    character_set_server=utf8
    datadir=/var/lib/mysql
    server-id=1000
    log-bin=/var/lib/mysql/mysql-bin
    binlog-do-db=heima

5.2 设置用户权限

添加一个仅用于数据同步的账户，出于安全考虑，这里仅提供对heima这个库的操作权限。

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

5.3 安装Canal

以docker容器为例

docker run -p 11111:11111 --name canal \
    -e canal.destinations=heima \
    -e canal.instance.master.address=mysql:3306  \
    -e canal.instance.dbUsername=canal  \
    -e canal.instance.dbPassword=canal  \
    -e canal.instance.connectionCharset=UTF-8 \
    -e canal.instance.tsdb.enable=true \
    -e canal.instance.gtidon=false  \
    -e canal.instance.filter.regex=heima\\..* \
    -d canal/canal-server:v1.1.5

说明:

- \`-p 11111:11111\`：这是canal的默认监听端口

- \`-e canal.destinations=heima\`：canal集群的名字,要和canal服务中设置的集群名字保持一致

- \`-e canal.instance.master.address=192.168.136.133:3306\`：数据库地址和端口

- \`-e canal.instance.dbUsername=canal\`：数据库用户名

- \`-e canal.instance.dbPassword=canal\` ：数据库密码

- \`-e canal.instance.tsdb.enable=true\` ：表示是否打开tsdb开关

- \`-e canal.instance.gtidon=false\` ：表示是否是GTID模式

- \`-e 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

\`\`\`

5.4 Canal客户端服务

① 客户端概述

Canal提供了各种语言的客户端，当Canal监听到binlog变化时，会通知Canal的客户端。

我们可以利用Canal提供的Java客户端，监听Canal通知消息。当收到变化的消息时，完成对数据的更新。

当然, 市面上有很多Canal客户端, 我们以GitHub上的第三方开源的canal-starter客户端, 并且结合SpringBoot为例, 进行演示。

② 引入依赖

<dependency>
        <groupId>top.javatool</groupId>
        <artifactId>canal-spring-boot-starter</artifactId>
        <version>1.2.1-RELEASE</version>
    </dependency>

③ 编写配置

canal:
      destination: heima # canal的集群名字，要与安装canal时设置的名称一致
      server: 192.168.136.135:11111 # canal服务地址

④ 实体类

package com.itheima.domain;
    
    import lombok.Data;
    
    @Data
    public class Hotel {
        private Long id;
        private String name;
        private String address;
        private Integer price;
        private Integer score;
        private String brand;
        private String city;
        private String starName;
        private String business;
        private String longitude;
        private String latitude;
        private String pic;
    }

⑤ 编写监听器

> 通过实现\`EntryHandler\`接口编写监听器，监听Canal消息。

> 注意两点：

> - 实现类通过\`@CanalTable("tb\_hotel")\`指定监听的表信息

> - EntryHandler的泛型是与表对应的实体类

package com.itheima.handle;
    
    import com.itheima.domain.Hotel;
    import org.springframework.stereotype.Component;
    import top.javatool.canal.client.annotation.CanalTable;
    import top.javatool.canal.client.handler.EntryHandler;
    
    @CanalTable("tb_hotel")
    @Component
    public class HotelHandler implements EntryHandler<Hotel> {
    
        /**
         * mysql中数据有新增时自动执行
         * @param hotel 新增的数据
         */
        @Override
        public void insert(Hotel hotel) {
            //把新增数据hotel,添加到ES即可
        }
    
        /**
         * mysql中数据有修改时自动执行
         * @param before 修改前的数据
         * @param after 修改后的数据
         */
        @Override
        public void update(Hotel before, Hotel after) {
            //把修改数据,更新到ES即可
        }
    
        /**
         * ysql中数据有删除时自动执行
         * @param hotel 要删除的数据
         */
        @Override
        public void delete(Hotel hotel) {
            //把要删除的数据hotel,从ES删除即可
        }
    }

6.补充

该模式不仅仅能实现mysql到es之间的数据同步. 它适用于任何从mysql到其他接受的数据同步。

[bb4b257695193c54d2d9de71a1b1b3f7.jpeg]: https://img-blog.csdnimg.cn/img_convert/bb4b257695193c54d2d9de71a1b1b3f7.jpeg
[8dcd25f9a245ec8aa94d1c3fd3d194d5.jpeg]: https://img-blog.csdnimg.cn/img_convert/8dcd25f9a245ec8aa94d1c3fd3d194d5.jpeg
[de0efdacd6561e8ac4bc4c7300f43964.jpeg]: https://img-blog.csdnimg.cn/img_convert/de0efdacd6561e8ac4bc4c7300f43964.jpeg
[2f0036f88bcf5e9a93953072e9ddbb0e.jpeg]: https://img-blog.csdnimg.cn/img_convert/2f0036f88bcf5e9a93953072e9ddbb0e.jpeg