Canal CDC-CFANZ编程社区

文章目录

1.Canal介绍

1.1 Mysql 的binlog介绍
1.2 Canal 的运行原理
1.3 Canal使用场景

2.Mysql 的配置准备
3.Canal 的准备
4.Canal 数据结构分析
5.Java 代码
6.Kafka 测试

1.Canal介绍

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

1.1 Mysql 的binlog介绍

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

binlog可分为STATEMENT, MIXED, ROW

statement：语句级，binlog 会记录每次一执行写操作的语句。相对 row 模式节省空间，但是可能产生不一致性，比如“update tt set create_date=now()”，如果用 binlog 日志进行恢复，由于执行时间不同可能产生的数据就不同。
row：行级， binlog 会记录每次操作后每行记录的变化。
mixed：statement 的升级版，一定程度上解决了，因为一些情况而造成的 statement模式不一致问题，默认还是 statement，在某些情况下譬如：当函数中包含 UUID() 时；包含AUTO_INCREMENT 字段的表被更新时；执行 INSERT DELAYED 语句时；用 UDF 时；会按照ROW 的方式进行处理

Canal 想做监控分析，选择 row 格式比较合适。

1.2 Canal 的运行原理

Canal CDC_java
Mysql的主从复制:

Master改变数据, 写入到二进制文件中
slave 从master 发送dump协议, 读取二进制文件到自己的relay log
slave读取relay log到自己的数据库

canal就是将自己伪装为slave

1.3 Canal使用场景

1> 进行异地数据库之间的同步框架

Canal CDC_大数据_02
2> 更新缓存, 实现缓存和数据库的一致性
Canal CDC_json_03
3> 抓取业务表的新增变化数据，用于制作实时统计（我们就是这种场景）

2.Mysql 的配置准备

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

vim /etc/my.cnf

Canal CDC_java_04

systemctl restart mysqld

3.Canal 的准备

Canal CDC_OLAP_05
修改配置
Canal CDC_OLAP_06

Canal CDC_mysql_07

Canal CDC_json_08
Canal CDC_json_09
防火墙关闭

Canal CDC_OLAP_10

4.Canal 数据结构分析

Canal CDC_java_11
发送的是Message, 由很多Entry组成, 一个Entry对应一个Sql命令

Entry: TableName, EntryTyple, StoreValue, RowChange

RowChange为反序列化后的数据, 如果要使用的话必须通过StoreValue反序列化为RowChange后才可以使用

5.Java 代码

maven

<dependency>
            <groupId>com.alibaba.ottergroupId>
            <artifactId>canal.clientartifactId>
            <version>1.0.25version>
        dependency>

public class Test {
    public static void main(String[] args)throws Exception {
        // Canal中的数据结构: Message - Entry(对应一个Sql) - TableName, EntryType, StoreValue-RowChange
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.66.66", 11111), "example", "", "");

        while (true){
            // 连接
            connector.connect();
            // 订阅
            connector.subscribe("cdc_test.*");
            // 获取数据
            Message message = connector.get(100);

            // 获取Entry 集合
            List<CanalEntry.Entry> entries = message.getEntries();

            if (entries.size() <= 0){
                System.out.println("稍等一会.........");
                Thread.sleep(1000);
            }else {
                for (CanalEntry.Entry entry : entries) {
                    // 1.获取表名
                    String tableName = entry.getHeader().getTableName();
                    // 2.获取类型
                    CanalEntry.EntryType entryType = entry.getEntryType();
                    // 3.获取序列化后的数据
                    ByteString storeValue = entry.getStoreValue();
                    // 判断当前的类型是否为Row
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)){
                        // 5.反序列化数据
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                        // 6.获取EventType
                        CanalEntry.EventType eventType = rowChange.getEventType();
                        // 7.获取数据集
                        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                        for (CanalEntry.RowData rowData : rowDatasList) {
                            JSONObject beforeJson = new JSONObject();
                            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                            for (CanalEntry.Column column : beforeColumnsList) {
                                beforeJson.put(column.getName(), column.getValue());
                            }

                            JSONObject afterJson = new JSONObject();
                            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                            for (CanalEntry.Column column : afterColumnsList) {
                                afterJson.put(column.getName(), column.getValue());
                            }
                            System.out.println("Table：" + tableName + "eventType: " + eventType + "before: " + beforeJson + "after: " + afterJson);


                        }
                    }else {
                        System.out.println("当前数据类型为：" + entryType);
                    }

                }
            }
        }
    }
}

Canal CDC_OLAP_12

6.Kafka 测试

修改 canal.properties 中 canal 的输出 model，默认 tcp，改为输出到 kafka

Canal CDC_json_13
修改 Kafka 集群的地址

canal.mq.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092

修改 instance.properties 输出到 Kafka 的主题以及分区数

# mq config
canal.mq.topic=canal_test
canal.mq.partitionsNum=1
# hash partition config
#canal.mq.partition=0
#canal.mq.partitionHash=mytest.person:id,mytest.role:id

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic canal_test

向数据库添加数据后

Canal CDC_mysql_14