mysql数据同步解决方案_数据同步解决方案-canal

喜欢ヅ旅行 2022-11-18 10:54 305阅读 0赞

**mysql数据同步解决方案\_数据同步解决方案-canal**

工作原理

canal 译意为水道，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。

\* MySQL主备复制原理

MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)

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

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

\* canal 工作原理

canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议

MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )

canal 解析 binary log 对象(原始为 byte 流)

mysql开启binlog

1. 登录mysql，使用       show variables like 'log\_bin';       命令查看是否已开启binlog

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1Y2FpZnUxOTg5_size_16_color_FFFFFF_t_70][]

2. 如果没有开启的话，则需要修改my.cnf文件指定binlog信息

\# 当前数据库唯一编号

server-id=12345

\# 二进制日志存储地址(mysql-bin是文件前缀)

log-bin=/var/lib/mysql/logs/mysql-bin

\# binlog日志格式，mysql默认采用statement，canal建议使用ROW

binlog\_format=ROW

\# binlog过期清理时间

expire\_logs\_days=7

\# binlog每个日志文件大小

max\_binlog\_size=100m

\# binlog缓存大小

binlog\_cache\_size=4m

\# 最大binlog缓存大小

max\_binlog\_cache\_size=512m

这里指定了log-bin=/var/lib/mysql/logs目录，logs是手动创建出来的，且要 chmod -R 777 logs 赋予权限，否则mysql启动失败

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1Y2FpZnUxOTg5_size_16_color_FFFFFF_t_70 1][]

3. 修改完成之后，重启mysqld的服务。

canal服务端安装

1. 下载地址

https://github.com/alibaba/canal/releases/tag/canal-1.0.24

2. 解压

mkdir canal && tar -zxvf canal.deployer-1.0.24.tar.gz -C canal

3. 修改pid (canal像一个从库去读取mysql数据，所以他们的id是不能冲突的，mysql的刚才改为12345了)

vim conf/canal.properties

![20210413101330894.png][]

4. 配置数据库信息和监听规则

\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#

\#\# mysql的id

canal.instance.mysql.slaveId = 12345

\# position info

canal.instance.master.address = 127.0.0.1:3306

canal.instance.master.journal.name =

canal.instance.master.position =

canal.instance.master.timestamp =

\#canal.instance.standby.address =

\#canal.instance.standby.journal.name =

\#canal.instance.standby.position =

\#canal.instance.standby.timestamp =

\# username/password

canal.instance.dbUsername = root

canal.instance.dbPassword = 123456

canal.instance.defaultDatabaseName =

canal.instance.connectionCharset = UTF-8

\#所有库的所有表

\#canal.instance.filter.regex = .\*\\\\..\*

\#meiye库的system\_user表

\#canal.instance.filter.regex =meiye.system\_user

\#meiye 和 brm 的所有表

canal.instance.filter.regex =meiye.\*,brm.\*

\# table black regex

canal.instance.filter.black.regex =

\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1Y2FpZnUxOTg5_size_16_color_FFFFFF_t_70 2][]

代码测试

1. 导入依赖

com.xpand

starter-canal

0.0.1-SNAPSHOT

2. 配置canal地址信息

canal:

client:

instances:

example:

host: 192.168.200.100

port: 11111

batchSize: 1000

3. 启动类开启canal注解

@EnableCanalClient //声明当前的服务是canal的客户端

4. 测试代码

import com.alibaba.otter.canal.protocol.CanalEntry;

import com.xpand.starter.canal.annotation.\*;

/\*\*

\* mysql数据监听

\*/

@CanalEventListener

public class CanalDataEventListener \{

/\*\*

\* 监听新增事件

\* @param eventType 操作事件类型(新增)

\* @param rowData 发生变化的一行数据

\*/

@InsertListenPoint

public void onEventInsert(CanalEntry.EventType eventType, CanalEntry.RowData rowData)\{

for (CanalEntry.Column column : rowData.getAfterColumnsList())\{

System.out.println("列名："+column.getName()+"----------变化的数据"+column.getValue());

/\*\*

\* 监听修改事件

\* @param eventType 操作事件类型(修改)

\* @param rowData 发生变化的一行数据

\*/

@UpdateListenPoint

public void onEventUpdate(CanalEntry.EventType eventType, CanalEntry.RowData rowData)\{

for (CanalEntry.Column column : rowData.getBeforeColumnsList())\{

System.out.println("修改前列名："+column.getName()+"----------变化的数据"+column.getValue());

for (CanalEntry.Column column : rowData.getAfterColumnsList())\{

System.out.println("修改后列名："+column.getName()+"----------变化的数据"+column.getValue());

/\*\*

\* 监听删除事件

\* @param eventType 操作事件类型(删除)

\* @param rowData 发生变化的一行数据

\*/

@DeleteListenPoint

public void onEventDelete(CanalEntry.EventType eventType, CanalEntry.RowData rowData)\{

for (CanalEntry.Column column : rowData.getBeforeColumnsList())\{

System.out.println("删除前的列名："+column.getName()+"----------变化的数据"+column.getValue());

/\*\*

\* 自定义监听

\*/

@ListenPoint(

eventType = \{CanalEntry.EventType.INSERT, CanalEntry.EventType.DELETE, CanalEntry.EventType.UPDATE\}, // 监听的事件类型

schema = \{"changgou\_content"\}, // 监听的库

table = \{"tb\_content"\} // 指定监控的表

)

public void onListenPoint(CanalEntry.EventType eventType, CanalEntry.RowData rowData) \{

for (CanalEntry.Column column : rowData.getBeforeColumnsList())\{

System.out.println("自定义操作前：列名："+column.getName()+"----------变化的数据"+column.getValue());

for (CanalEntry.Column column : rowData.getAfterColumnsList())\{

System.out.println("自定义操作后：列名："+column.getName()+"----------变化的数据"+column.getValue());

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1Y2FpZnUxOTg5_size_16_color_FFFFFF_t_70 3][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1Y2FpZnUxOTg5_size_16_color_FFFFFF_t_70]: /images/20221022/0eea87c2ac93475dbe5d4c39fb75feeb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1Y2FpZnUxOTg5_size_16_color_FFFFFF_t_70 1]: /images/20221022/d8e0e2ab26b245be99d0867cb7c6ce36.png
[20210413101330894.png]: /images/20221022/fd75e0e2706a4ddcaf1e3a35621b1295.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1Y2FpZnUxOTg5_size_16_color_FFFFFF_t_70 2]: /images/20221022/73d2d07fdeae40fa9be2b45415727711.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1Y2FpZnUxOTg5_size_16_color_FFFFFF_t_70 3]: /images/20221022/c67025e9b26c4aa4a0ca7f1005741a03.png