canal同步mysql数据到es、oracle、mq、redis和mysql中

布满荆棘的人生 2021-08-19 11:41 771阅读 0赞

# 前言： #

mysql数据的同步只能解决数据结构相同的数据同步，假如我想把mysql中的数据变更同步到oracle、es、mq、redis或者说同样是mysql但是表结构不同的数据库中，mysql自带的主从功能已经满足不了这种需求了；阿里巴巴推出了一个开源框架[canal][] 就能解决这类问题。

# 环境： #

1、canal.deployer：1.1.4  
2、canal.adapter：1.1.4  
3、elasticsearch：6.7.2 （1.1.4canal不支持7.x的es，支持的es版本为 6.4.3-6.8）  
4、mysql：5.7  
5、jdk：1.8

注意：本文主要是讲解canal，所以默认已经有一点es基础，并且es、es的可视化工具和mysql已经安装好了。。。

# 准备操作： #

### 一、canal安装： ###

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

说明：这里需要下载的是：  
1、canal.deployer1.1.4版本- - - 可以理解为相当于canal的服务端  
2、canal.adapter1.1.4版本- - - 可以理解为相当于canal的插件  
3、最新的是1.1.5版本，但是是快照版，我们这里还是选择稳定版本

### 二、mysql修改 ###

**1、配置文件**  
\[mysqld\]  
log-bin=mysql-bin \#开启 binlog  
binlog-format=ROW \# 选择 ROW 模式（[不了解的可以看下博主之前的文章][Link 1]）  
server\_id=1 \# 不要和 canal 的 slaveId 重复就行

注意：  
1、mac下还需要修改：bind-address = 127.0.0.1->bind-address = 0.0.0.0  
2、修改完配置文件一定要记得重启mysql

**2、创建一个可以远程连接的用户**

CREATE USER canal IDENTIFIED BY 'canal';  
    GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
    FLUSH PRIVILEGES;

### 三、canal-deployer修改： ###

**1、打开deployer/conf/example**

**2、删除除instance.properties以外的所有东西（如果不是新下载的）**

**3、修改配置文件：instance.properties：**

## mysql serverId , v1.0.26+ will autoGen   
    # 这个东西可以不设置，如果要设置别和上面mysql配置文件中的值重复就行
    # canal.instance.mysql.slaveId=0

canal.instance.master.address=mysql启动的ip:端口  #例如：192.168.34.66:3306

# username/password
    canal.instance.dbUsername=canal   # 刚才自己创建账号
    canal.instance.dbPassword=canal   # 刚才自己创建密码
    canal.instance.connectionCharset = UTF-8
    # enable druid Decrypt database password

# 这个可以根据需要修改过滤，默认是直接监听所有
    canal.instance.filter.regex=.*\\..*    #例如：canal_tsdb.test_table1

### 四、启动canal-deployer ###

linux/mac：切换到deployer/bin目录下 sh startup.sh 即可  
windows：切换到deployer/bin目录下双击startup.bat 即可

# 数据同步一：java代码实现 #

**1、创建一个maven工程，添加以下依赖**

<dependency>
                <groupId>com.alibaba.otter</groupId>
                <artifactId>canal.client</artifactId>
                <version>1.1.4</version>
            </dependency>

**2、编写java代码，获取mysql中数据的变化（这段代码canal的github上有）**

import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.client.CanalConnectors;
    import com.alibaba.otter.canal.protocol.CanalEntry.*;
    import com.alibaba.otter.canal.protocol.Message;
    import java.net.InetSocketAddress;
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.List;
    
    
    public class SimpleCanalClientExample {
        public static void main(String args[]) {
            // 创建链接
            //192.168.35.254为canal-deployer启动的ip地址
            //canal-deployer如果你没有修改过的话，默认端口是11111
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.35.254", 11111), "example", "", "");
            //底层默认是1000
            int batchSize = 1;
            try {
                connector.connect();
                connector.subscribe(".*\\..*");
                // connector.subscribe("canal_tsdb.test_table1");
                //把上次停止后未提交的数据回滚，因为不确定是否已处理
                connector.rollback();
                while (true) {
                    System.out.println("当前时间为：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                    //每次获取多少条数据
                    Message message = connector.getWithoutAck(batchSize);
                    //System.out.println("内容为："+ JSON.toJSONString(message));
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        try {
                            Thread.sleep(10000);
                        } catch (InterruptedException e) {
                        }
                    } else {
                        try {
                            System.out.println(batchId);
                            printEntry(message.getEntries());
                            // int i=1/0;
                            connector.ack(batchId); // 提交确认
                        } catch (Exception e) {
                            connector.rollback(batchId); // 处理失败, 回滚数据
                        }
                    }
                }
            } finally {
                connector.disconnect();
            }
        }
        private static void printEntry(List<Entry> entrys) {
            System.out.println("读取长度为：" + entrys.size());
            for (Entry entry : entrys) {
                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                    continue;
                }
                RowChange rowChage = null;
                try {
                    rowChage = RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                            e);
                }
                EventType eventType = rowChage.getEventType();
                System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                        eventType));
                for (RowData rowData : rowChage.getRowDatasList()) {
                    if (eventType == EventType.DELETE) {
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == EventType.INSERT) {
                        printColumn(rowData.getAfterColumnsList());
                    } else {
                        System.out.println("-------> before");
                        printColumn(rowData.getBeforeColumnsList());
                        System.out.println("-------> after");
                        printColumn(rowData.getAfterColumnsList());
                    }
                }
            }
        }
        private static void printColumn(List<Column> columns) {
            for (Column column : columns) {
                System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            }
        }
    }

**3、运行之后，只要我们修改mysql中的数据，就会有东西打印出来，如图所示：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3R1ZXNkYXltYQ_size_16_color_FFFFFF_t_70]  
**4、分析：** 上图可以看到，其实在java代码中已经能拿到mysql中数据的变化，可以具体到那个数据源，那张表，那条记录，更新的甚至都能知道更新前和更新后分别是什么

**5、思考：** 个人觉得这种方式比较灵活，在java代码中拿到了数据源，表，字段，具体什么操作等信息之后，如果想同步数据，不是分分钟的事情，无论你是es，oracle，mq，redis，mysql还是什么hbase，只要java能连的东西，都能同步数据，但是这种方式同样也有缺点，就是主库只要一发生变化，从库就要执行相同的操作，而且是用java去操作对应的备份单元，这样就要求从库和主库的性能不能差太多。。。

# 数据同步二：mysql同步到myql中 #

**1、准备两个mysql数据库**

**2、主从库的账号密码都得保证能远程连接**

**3、注意：** mysql版本不能是8.x，目前测出来，canal1.1.4不支持mysql8.x

**4、修改canal-adapter/conf/application.yml配置文件：**

canal.conf:
      mode: tcp
     # 修改成自己canal-deployer启动的ip和端口
      canalServerHost: 192.168.35.254:11111
      batchSize: 500
      syncBatchSize: 1000
      retries: 0
      timeout:
      accessKey:
      secretKey:
      srcDataSources:
        defaultDS:
          # 换成自己mysql主库的地址和数据源
          url: jdbc:mysql://192.168.34.66:3306/canal_tsdb?useUnicode=true
          username: canal  # 换成自己mysql主库的账号
          password: canal  #  换成自己mysql主库的密码
      canalAdapters:
      - instance: example
        groups:
        - groupId: g1
          outerAdapters:
           - name: logger
           - name: rdb
             key: mysql1
             properties:
               jdbc.driverClassName: com.mysql.jdbc.Driver
               jdbc.url: jdbc:mysql://从库ip:3306/从库数据源?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true
               jdbc.username: 从库账号
               jdbc.password: 从库密码

**5、修改canal-adapter/conf/rdb/mytest\_user.yml配置文件：**

**6、启动canal-adapter：**

linux/mac：切换到adapter/bin目录下 sh startup.sh 即可  
windows：切换到adapter/bin目录下双击startup.bat 即可

**7、效果：** 主库canal\_tsdb下的test1表和从库下bqorderdb下的test2表实现了同步

**8、注意：**

1、canal-adapter/conf/rdb/mytest\_user.yml配置文件修改的时候，需要注意主库和从库之间的表结构的问题，如果相同，可以直接配置mapAll: true，如果不一样，则需要配置targetColumns，格式为 从表字段名字: 主表字段名字

2、canal-adapter/conf/rdb/mytest\_user.yml的groupId应该和canal-adapter/conf/application.yml中的保持一致

**9、思考：** 如果需要同时同步多张表怎么办，或者说来自不同数据源的表？

**10、解决：** 在canal-adapter/conf/rdb/mytest\_user.yml和canal-adapter/conf/application.yml中再增加一个groupId即可

canal.conf:
      mode: tcp
      canalServerHost: 192.168.35.254:11111
      batchSize: 500
      syncBatchSize: 1000
      retries: 0
      timeout:
      accessKey:
      secretKey:
      srcDataSources:
        defaultDS:
          url: jdbc:mysql://192.168.34.66:3306/canal_tsdb?useUnicode=true
          username: canal
          password: canal
      canalAdapters:
      - instance: example
        groups:
        - groupId: g1
          outerAdapters:
           - name: logger
           - name: rdb
             key: mysql1
             properties:
               jdbc.driverClassName: com.mysql.jdbc.Driver
               jdbc.url: jdbc:mysql://从库ip:3306/从库数据源?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true
               jdbc.username: 从库账号
               jdbc.password: 从库密码
        - groupId: g2
          outerAdapters:
           - name: logger
           - name: rdb
             key: mysql1
             properties:
               jdbc.driverClassName: com.mysql.jdbc.Driver
               jdbc.url: jdbc:mysql://从库ip:3306/从库数据源?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true
               jdbc.username: 从库账号
               jdbc.password: 从库密码

dataSourceKey: defaultDS
    destination: example
    groupId: g1
    outerAdapterKey: mysql1
    concurrent: true
    dbMapping:
       #配置数据库--主库
       database: canal_tsdb
       #配置表--主库
       table: test1
       #从库
       targetTable: bqorderdb.test2
       targetPk:
         idid: id
       #如果两张表的结构是一样的话，直接设置为true
       #mapAll: true
       targetColumns: 
       #从表字段名字: 主表字段名字
        idid: id
        name: name
        address: address
    groupId: g2
    outerAdapterKey: mysql1
    concurrent: true
    dbMapping:
       #配置数据库--主库
       database: canal_tsdb
       #配置表--主库
       table: canal_test
       #从库
       targetTable: bqorderdb.test3
       targetPk:
         idid: id
       #如果两张表的结构是一样的话，直接设置为true
       #mapAll: true
       targetColumns: 
       #从表字段名字: 主表字段名字
        idid: id
        name: name
        address: address

# 数据同步三：mysql同步到es中 #

**1、准备一个mysql数据库**

**2、准备一个es，但是版本不能低于6.4.3也不能高于6.8**

**3、创建一个index、type和mappings**

**4、修改canal-adapter/conf/application.yml配置文件：**

**5、修改canal-adapter/conf/test1.yml配置文件：**

dataSourceKey: defaultDS
    destination: example
    groupId: g1
    esMapping:
      _index: test1
      _type: _doc
      _id: _id
      upsert: true
      sql: "select a.id as _id,a.name,a.address from canal_tets a"
      commitBatch: 3000

**6、注意：**

1、canal-adapter/conf/rdb/mytest\_user.yml的groupId应该和canal-adapter/conf/application.yml中的保持一致

2、在创建es的index的时候，如果是7以下的版本，还是需要创建type的，这里用的es版本是6.7.2，所以也是需要创建的

3、es同步的主要关键在于这句sql，他能把mysql表中的字段映射到es的field上，要映射几个字段就查几个字段

**7、效果：** 我们可以看到mysql中操作一条数据，es中也会对应的变更  
![在这里插入图片描述][20191129094513862.jpg]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3R1ZXNkYXltYQ_size_16_color_FFFFFF_t_70 1]

[canal]: https://github.com/alibaba/canal
[Link 1]: https://blog.csdn.net/tuesdayma/article/details/80545434
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3R1ZXNkYXltYQ_size_16_color_FFFFFF_t_70]: /images/20210511/226bdacb35d54a59b9de901988c52de9.png
[20191129094513862.jpg]: /images/20210511/d2ebc09ea7784d52be40e698d1060bb1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3R1ZXNkYXltYQ_size_16_color_FFFFFF_t_70 1]: /images/20210511/31cd2a27c3b84eeb8474d098d43f0777.png