目录

一、canal是什么

二、canal的原理

三、canal初尝

---

一、canal是什么

canal [kə'næl]，译意为水道/管道/沟渠，是阿里开源的一个增量数据变更收集的工具，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费的一种中间件。

二、canal的原理

说到canal的原理我们要先说明mysql的主从同步

2.1 mysql的主从同步：

（1）Master 主库将改变记录，写到binary log中

（2）Slave 从库向 MySQL Master 发送 dump 协议，将 Master 主库的 binary log events 拷到它的relay log

（3）Slave 从库读取并重做relay log中的事件，将改变的数据同步到自己的数据库

其他：

（1）在线上环境，对于读远大于写的场景，一般都是一主多从，主库进行写操作，从库进行读操作，然后主库内容更新再同步给从库

（2）binary log主要作⽤是记录数据库中表的更改，它只记录改变数据的sql，不改变数据的sql不会写⼊，⽐如select语句⼀般不会被记录，因为他们不会对数据产⽣任何改动

（3）relay log的结构和binlog非常相似，只不过他多了一个master.info和relay-log.info的文件，master.info记录了上一次读取到master同步过来的binlog的位置，以及连接master和启动复制必须的所有信息。relay-log.info记录了文件复制的进度，下一个事件从什么位置开始，由sql线程负责更新

2.2 canal的原理

canal其实本质就是将自己伪装成一个slave，同步主库的binary log

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

（2）mysql master 收到 dump 请求，开始推送 binary log 给 slave (也就是 canal)

（3）canal 解析 binary log 对象 

三、canal初尝

数据库中表的更改都记录在binlog日志中，但是binlog日志也有三种格式，我们可以根据自己的需要决定到底使用哪一种，这边我们为了便于观察，使用了row格式。

binlog格式	具体含义	优点	缺点
STATEMENT	语句级别，记录每一次执行写操作的语句，相对于ROW模式节省了空间，但是可能产生数据不一致如update tt set create_date=now()，由于执行时间不同产生的数据就不同	节省空间	可能造成数据不一致
ROW	行级，记录每次操作后每行记录的变化。假如一个update的sql执行结果是1万行，statement只存一条，如果是row的话会把这个10000行的结果存这。	持数据的绝对一致性。因为不管sql是什么，引用了什么函数，他只记录执行后的效果	占用较大空间
MIXED	是statement的一种升级，由mysql server层智能选择是实用statement还是row，但是这种只能并不能保证百分之百正确	节省空间，同时兼顾了一定的一致性	还有些极个别情况依旧会造成不一致，另外statement和mixed对于需要对binlog的监控的情况都不方便

3.1 使用statement格式，对库里的4条数据进行update操作，position移动了340

# statement格式下详细的binlog日志记录，从position1792-2132
# The proper term is pseudo_replica_mode, but we use this compatibility alias
# to make the statement usable on server versions 8.0.24 and older.
/*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=1*/;
/*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
DELIMITER /*!*/;
# at 156
#230517 17:14:08 server id 1  end_log_pos 125 CRC32 0x34e0ed85  Start: binlog v 4, server v 8.0.26 created 230517 17:14:08 at startup
# Warning: this binlog is either in use or was not closed properly.
ROLLBACK/*!*/;
BINLOG '
4JpkZA8BAAAAeQAAAH0AAAABAAQAOC4wLjI2AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
AAAAAAAAAAAAAAAAAADgmmRkEwANAAgAAAAABAAEAAAAYQAEGggAAAAICAgCAAAACgoKKioAEjQA
CigBhe3gNA==
'/*!*/;
# at 1792
#230524 10:46:23 server id 1  end_log_pos 1871 CRC32 0xf198682b     Anonymous_GTID  last_committed=5    sequence_number=6   rbr_only=no original_committed_timestamp=1684896383380934   immediate_commit_timestamp=1684896383380934 transaction_length=340
# original_commit_timestamp=1684896383380934 (2023-05-24 10:46:23.380934 CST)
# immediate_commit_timestamp=1684896383380934 (2023-05-24 10:46:23.380934 CST)
/*!80001 SET @@session.original_commit_timestamp=1684896383380934*//*!*/;
/*!80014 SET @@session.original_server_version=80026*//*!*/;
/*!80014 SET @@session.immediate_server_version=80026*//*!*/;
SET @@SESSION.GTID_NEXT= 'ANONYMOUS'/*!*/;
# at 1871
#230524 10:46:23 server id 1  end_log_pos 1964 CRC32 0x9cfbf3c0     Query   thread_id=12    exec_time=0 error_code=0
SET TIMESTAMP=1684896383/*!*/;
SET @@session.pseudo_thread_id=12/*!*/;
SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=0, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
SET @@session.sql_mode=1168113696/*!*/;
SET @@session.auto_increment_increment=1, @@session.auto_increment_offset=1/*!*/;
/*!\C utf8mb4 *//*!*/;
SET @@session.character_set_client=255,@@session.collation_connection=255,@@session.collation_server=255/*!*/;
SET @@session.lc_time_names=0/*!*/;
SET @@session.collation_database=DEFAULT/*!*/;
/*!80011 SET @@session.default_collation_for_utf8mb4=255*//*!*/;
BEGIN
/*!*/;
# at 1964
#230524 10:46:23 server id 1  end_log_pos 2101 CRC32 0x723fdb20     Query   thread_id=12    exec_time=0 error_code=0
use `zhou1`/*!*/;
SET TIMESTAMP=1684896383/*!*/;
update player_role set player_name='zhouluying10'
/*!*/;
# at 2101
#230524 10:46:23 server id 1  end_log_pos 2132 CRC32 0x724686a9     Xid = 1480
COMMIT/*!*/;

3.2 实用row格式，对库里面的4条数据进行update操作，position移动了424

 

# row格式下详细的binlog日志记录，从position2132-2556
# The proper term is pseudo_replica_mode, but we use this compatibility alias
# to make the statement usable on server versions 8.0.24 and older.
/*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=1*/;
/*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
DELIMITER /*!*/;
# at 156
#230517 17:14:08 server id 1  end_log_pos 125 CRC32 0x34e0ed85  Start: binlog v 4, server v 8.0.26 created 230517 17:14:08 at startup
# Warning: this binlog is either in use or was not closed properly.
ROLLBACK/*!*/;
BINLOG '
4JpkZA8BAAAAeQAAAH0AAAABAAQAOC4wLjI2AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
AAAAAAAAAAAAAAAAAADgmmRkEwANAAgAAAAABAAEAAAAYQAEGggAAAAICAgCAAAACgoKKioAEjQA
CigBhe3gNA==
'/*!*/;
# at 2132
#230524 10:47:13 server id 1  end_log_pos 2211 CRC32 0x162f1f81     Anonymous_GTID  last_committed=6    sequence_number=7   rbr_only=yes    original_committed_timestamp=1684896433200961   immediate_commit_timestamp=1684896433200961 transaction_length=424
/*!50718 SET TRANSACTION ISOLATION LEVEL READ COMMITTED*//*!*/;
# original_commit_timestamp=1684896433200961 (2023-05-24 10:47:13.200961 CST)
# immediate_commit_timestamp=1684896433200961 (2023-05-24 10:47:13.200961 CST)
/*!80001 SET @@session.original_commit_timestamp=1684896433200961*//*!*/;
/*!80014 SET @@session.original_server_version=80026*//*!*/;
/*!80014 SET @@session.immediate_server_version=80026*//*!*/;
SET @@SESSION.GTID_NEXT= 'ANONYMOUS'/*!*/;
# at 2211
#230524 10:47:13 server id 1  end_log_pos 2296 CRC32 0x18e30c17     Query   thread_id=12    exec_time=0 error_code=0
SET TIMESTAMP=1684896433/*!*/;
SET @@session.pseudo_thread_id=12/*!*/;
SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=0, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
SET @@session.sql_mode=1168113696/*!*/;
SET @@session.auto_increment_increment=1, @@session.auto_increment_offset=1/*!*/;
/*!\C utf8mb4 *//*!*/;
SET @@session.character_set_client=255,@@session.collation_connection=255,@@session.collation_server=255/*!*/;
SET @@session.lc_time_names=0/*!*/;
SET @@session.collation_database=DEFAULT/*!*/;
/*!80011 SET @@session.default_collation_for_utf8mb4=255*//*!*/;
BEGIN
/*!*/;
# at 2296
#230524 10:47:13 server id 1  end_log_pos 2361 CRC32 0x9666ea38     Table_map: `zhou1`.`player_role` mapped to number 87
# at 2361
#230524 10:47:13 server id 1  end_log_pos 2525 CRC32 0x1e5c12f2     Update_rows: table id 87 flags: STMT_END_F
 
BINLOG '
sXptZBMBAAAAQQAAADkJAAAAAFcAAAAAAAEABXpob3UxAAtwbGF5ZXJfcm9sZQACDw8EgACAAAAC
A/z/ADjqZpY=
sXptZB8BAAAApAAAAN0JAAAAAFcAAAAAAAEAAgAC//8AATEMemhvdWx1eWluZzEwAAExDHpob3Vs
dXlpbmcxMQABMgx6aG91bHV5aW5nMTAAATIMemhvdWx1eWluZzExAAEzDHpob3VsdXlpbmcxMAAB
Mwx6aG91bHV5aW5nMTEAATQMemhvdWx1eWluZzEwAAE0DHpob3VsdXlpbmcxMfISXB4=
'/*!*/;
### UPDATE `zhou1`.`player_role`
### WHERE
###   @1='1' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
###   @2='zhouluying10' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
### SET
###   @1='1' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
###   @2='zhouluying11' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
### UPDATE `zhou1`.`player_role`
### WHERE
###   @1='2' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
###   @2='zhouluying10' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
### SET
###   @1='2' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
###   @2='zhouluying11' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
### UPDATE `zhou1`.`player_role`
### WHERE
###   @1='3' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
###   @2='zhouluying10' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
### SET
###   @1='3' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
###   @2='zhouluying11' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
### UPDATE `zhou1`.`player_role`
### WHERE
###   @1='4' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
###   @2='zhouluying10' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
### SET
###   @1='4' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
###   @2='zhouluying11' /* VARSTRING(128) meta=128 nullable=0 is_null=0 */
# at 2525
#230524 10:47:13 server id 1  end_log_pos 2556 CRC32 0x92f9a85a     Xid = 1593
COMMIT/*!*/;
SET @@SESSION.GTID_NEXT= 'AUTOMATIC' /* added by mysqlbinlog */ /*!*/;
DELIMITER ;
# End of log file
/*!50003 SET COMPLETION_TYPE=@OLD_COMPLETION_TYPE*/;
/*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=0*/;

show variables like 'binlog_format'; // 查看相关的binlog的使用格式 show master status; // 查看当前binlog的状态 set session binlog_format='statement'; // 修改为statement show binlog events in 'mysql-bin.000002'; // 查看粗略的binlog日志 mysqlbinlog -vv /usr/local/mysql/data/binlog.000007 --start-position=1792 // 查看详细的从某个游标的binlog日志

3.3 row格式的binlog样式，包含

（1）当前变更的sql是语句类型

（2）操作的数据库 database

（3）操作的表 table

（4）变更前后的数据

• insert语句相关的投递消息的消息体

// insert into player_role values(4,'zhouluying4');
{
    "data":[
        {
            "player_id":"4",
            "player_name":"zhouluying4"
        }
    ],
    "database":"zhou1",
    "es":1683267248000,
    "gtid":"",
    "id":2,
    "isDdl":false,
    "mysqlType":{
        "player_id":"varchar(32)",
        "player_name":"varchar(32)"
    },
    "old":null,
    "pkNames":[
        "player_id"
    ],
    "sql":"",
    "sqlType":{
        "player_id":12,
        "player_name":12
    },
    "table":"player_role",
    "ts":1683267249020,
    "type":"INSERT"
}

• update语句相关的投递消息的消息体

// update player_role set player_name='zhouluying5' where player_id=4; 
{
    "data":[
        {
            "player_id":"4",
            "player_name":"zhouluying5"
        }
    ],
    "database":"zhou1",
    "es":1683267310000,
    "gtid":"",
    "id":3,
    "isDdl":false,
    "mysqlType":{
        "player_id":"varchar(32)",
        "player_name":"varchar(32)"
    },
    "old":[
        {
            "player_name":"zhouluying4"
        }
    ],
    "pkNames":[
        "player_id"
    ],
    "sql":"",
    "sqlType":{
        "player_id":12,
        "player_name":12
    },
    "table":"player_role",
    "ts":1683267310753,
    "type":"UPDATE"
}

• delete语句相关的投递消息的消息体

// delete from player_role where player_id=4;
{
    "data":[
        {
            "player_id":"4",
            "player_name":"zhouluying5"
        }
    ],
    "database":"zhou1",
    "es":1683267383000,
    "gtid":"",
    "id":4,
    "isDdl":false,
    "mysqlType":{
        "player_id":"varchar(32)",
        "player_name":"varchar(32)"
    },
    "old":null,
    "pkNames":[
        "player_id"
    ],
    "sql":"",
    "sqlType":{
        "player_id":12,
        "player_name":12
    },
    "table":"player_role",
    "ts":1683267383843,
    "type":"DELETE"
}

从上面我们可以观察到，INSERT和DELETE语句产生的binary log中没有相关的old对象是一个null值，但是UPDATE是有相关具体的旧的数据值的