Canal 数据同步

Canal基本介绍

什么是Canal

阿里巴巴 B2B 公司，因为业务的特性，卖家主要集中在国内，买家主要集中在国外，所以衍生出了同步杭州和美国异地机房的需求，从 2010 年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务。Canal 是用 Java 开发的基于数据库增量日志解析，提供增量数据订阅&消费的中间件。目前。Canal 主要支持了 MySQL 的 Binlog 解析，解析完成后才利用 Canal Client 来处理获得的相关数据。（数据库同步需要阿里的 Otter 中间件，基于 Canal）。

MySQL的Binlog

MySQL 的二进制日志可以说 MySQL 最重要的日志了，它记录了所有的 DDL 和 DML(除了数据查询语句)语句，以事件形式记录，还包含语句所执行的消耗的时间，MySQL 的二进制日志是事务安全型的。

一般来说开启二进制日志大概会有 1%的性能损耗。二进制有两个最重要的使用场景:

其一：MySQL Replication 在 Master 端开启 Binlog，Master 把它的二进制日志传递给 Slaves来达到 Master-Slave 数据一致的目的。

其二：自然就是数据恢复了，通过使用 MySQL Binlog 工具来使恢复数据。二进制日志包括两类文件：二进制日志索引文件（文件名后缀为.index）用于记录所有的二进制文件，二进制日志文件（文件名后缀为.00000*）记录数据库所有的 DDL 和 DML(除了数据查询语句)语句事件。

Canal 的工作原理

很简单，就是把自己伪装成 Slave，假装从 Master 复制数据。

使用场景

1）原始场景： 阿里 Otter 中间件的一部分

Otter 是阿里用于进行异地数据库之间的同步框架，Canal 是其中一部分。

2）常见场景 1：更新缓存

3）常见场景 2：抓取业务表的新增变化数据，用于制作实时统计（我们就是这种场景）

数据库准备

创建数据库

创建数据表

CREATE TABLE user_info(
`id` VARCHAR(255),
`name` VARCHAR(255),
`sex` VARCHAR(255)
);

修改配置文件开启 Binlog

[atguigu@hadoop102 module] vim /etc/my.cnf
server-id=1
log-bin=mysql-bin
binlog_format=row
binlog-do-db=gmall

#测试是否开启成功
#验证
show variables like 'log_bin%';       #log_bin ON
show variables like 'binlog_format%'; #binlog_format ROW

注意：binlog-do-db 根据自己的情况进行修改，指定具体要同步的数据库，如果不配置 则表示所有数据库均开启 Binlog

重启 MySQL 使配置生效

systemctl restart mysqld

赋权限

在 MySQL 中执行

mysql> set global validate_password_length=4;
mysql> set global validate_password_policy=0;
mysql>  GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal' ;

Canal下载安装

#因为Canal默认是没有安装路径的，所以解压到改文件夹下
[root@yuan home]# mkdir -p /home/canal
#解压文件到/home/canal
[root@yuan home]# tar -zxvf canal.deployer-1.1.2.tar.gz -C  /home/canal

修改 canal.properties 的配置 canal.serverMode 为 tcp

[atguigu@hadoop102 conf]$ pwd
/opt/module/canal/conf
[atguigu@hadoop102 conf]$ vim canal.properties
#################################################
######### common argument ############# 
#################################################
canal.id = 1   
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =

canal.zookeeper.flush.period = 1000
canal.withoutNetty = false

canal.serverMode = tcp     #默认就是tcp

说明：

这个文件是 canal 的基本通用配置，canal 端口号默认就是 11111，修改 canal 的

输出 model，默认 tcp，改为输出到 kafka

多实例配置如果创建多个实例，通过前面 canal 架构，我们可以知道，一个 canal 服务

中可以有多个 instance，conf/下的每一个 example 即是一个实例，每个实例下面都有独立的

配置文件。默认只有一个实例 example，如果需要多个实例处理不同的 MySQL 数据的话，直

接拷贝出多个 example，并对其重新命名，命名和配置文件中指定的名称一致，然后修改

canal.properties 中的 canal.destinations=实例 1，实例 2，实例 3。

#################################################
######### destinations ############# 
#################################################
canal.destinations = example

修改 instance.properties

我们这里只读取一个 MySQL 数据，所以只有一个实例，这个实例的配置文件在conf/example 目录下

实时监控测试

创建Maven或者Spring Boot项目

 <!--client 和自己下载的版本一致即可-->
        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.client</artifactId>
            <version>1.1.2</version>
        </dependency>
        <!--kafka 和自己下载的版本一致即可-->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.4.1</version>
        </dependency>

通用监视类 –CanalClient

package com.cananl;
import com.alibaba.fastjson.JSONObject;
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 com.google.protobuf.ByteString;
import java.net.InetSocketAddress;
import java.util.List;

public class Test {
    public static void main(String[] args) throws Exception {
        // 获取连接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("182.61.41.88", 11111),
                "example", "", "");
        while (true) {
            //连接
            connector.connect();
            //订阅数据库 ，就是监视的是哪个数据库，跟上一个.*
            connector.subscribe("gmall.*");
            //获取数据，这里指的是单次拉取100条的SQL记录
            Message message = connector.get(100);
            //获取Entry集合
            List<CanalEntry.Entry> entries = message.getEntries();
            //判断集合是否为空，如果为空，则继续拉取数据
            if (entries.size() <= 0) {
                System.out.println("当前没有数据，等一会....");
                Thread.sleep(1000);
            }else {
                //遍历entries，单条解析
                for (CanalEntry.Entry entry : entries) {
                    //获取表名
                    String tableName = entry.getHeader().getTableName();
                    //获取类型
                    CanalEntry.EntryType entryType = entry.getEntryType();
                    //获取序列化后的数据
                    ByteString storeValue = entry.getStoreValue();
                    //判断当前entryType类型是否为ROWDADA
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)){
                        //反序列化数据
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                        //获取当前时间的操作类型
                        CanalEntry.EventType eventType = rowChange.getEventType();
                        //获取数据集
                        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                        //遍历RowDataList，并且打印数据集
                        for (CanalEntry.RowData rowData : rowDatasList) {
                            //获取更新前和后的数据
                            //获取更新前的数据
                            JSONObject beforData = new JSONObject();
                            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                            for (CanalEntry.Column column : beforeColumnsList) {
                                beforData.put(column.getName(),column.getValue());
                            }
                            //获取更新后的数据
                            JSONObject afterData= new JSONObject();
                            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                            for (CanalEntry.Column column : afterColumnsList) {
                                afterData.put(column.getName(),column.getValue());
                            }
                            //打印数据
                            System.out.println(
                                    "Table"+tableName+
                                            "EventType"+entryType+
                                            "beforData"+beforData+
                                            "afterData"+afterData);
                        }
                    }else {
                        System.out.println("当前操作类型为"+entryType);
                    }
                }
            }
        }
    }
}