1.前言

在我们系统开发过程中，根据业务场景很多数据库数据并不会直接给用户访问的，需要同步保存到ElasticSearch、Redis等存储应用当中（例如最常见的是搜索页面的ElasticSearch数据）。而阿里开源的框架Canal就是做这方面的功能，它可以把数据库（暂时只支持MySQL和Oracle部分版本）日志解析获取增量变更同步到其他存储应用去。

2.什么是Canal?

官网介绍，canal [kə’næl]，译意为水道/管道/沟渠，主要用途是基于MySQL数据库增量日志解析，提供增量数据订阅和消费。
从上述介绍我们可以简单认为Canal就是一个简单的增量数据同步工具。

2.1MySQL主备复制原理

根据官网介绍，MySQL主备复制原理如下：
 （1）Canal入门-风君雪科技博客
●MySQL master(主库)将数据变更（增删改）写入二进制日志(binary log，其中记录叫做二进制日志事件binary log events，可以通过show binlog events进行查看)。
●MySQL slave(从库)将master的binary log events 拷贝到它的中继日志(relay log)。
●MySQL slave(从库)重放relay log中事件，将数据变更反映它自己的数据。

2.2Canal工作原理

根据官网介绍，Canal工作原理如下图所示：
 （1）Canal入门-风君雪科技博客
●canal模拟MySQL slave的交互协议，伪装自己为MySQL slave，向MySQL master发送dump协议。
●MySQL master收到dump请求，开始推送binary log给slave(即canal )。
●canal解析binary log对象(原始为byte流)，再推送到MySQL、kafka、ElasticSearch等存储应用当中。

3.Canal能做什么？

早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务trigger（触发器）获取增量变更。从2010年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。所以Canal就是在这个场景中诞生的，它主要作用就是解决基于日志增量订阅和消费的业务，例如：
●数据库镜像。
●数据库实时备份。
●索引构建和实时维护(拆分异构索引、倒排索引等)。
●业务缓存刷新。
●带业务逻辑的增量数据处理（例如ElasticSearch、Redis数据同步）。
在我做过的项目中，cancal经常被用到如下场景：
●根据数据库的数据变更实时更新搜索引擎数据，比如我司电商场景下物料数据发生变更（例如后台上传更新物料信息、价格），实时同步到搜索引擎Elasticsearch上。
●根据数据库的数据变更实时更新缓存，比如专门运营人员每次修改物料品牌信息同时都会同步到Redis上。
●根据数据库的数据变更实时推送到消息队列，比如为了丰富自身系统物料库存，定时作业拉取第三方渠道物料库存推送到RabbitMQ等消息队列去消费入库。

4.如何搭建Canal？

4.1首先得安装个MySQL数据库

如果已经安装好MySQL数据库的，这一步可以跳过，如果没有安装好，请自行安装（也可以查看我之前写过一篇MySQL安装教程，不过个人建议最好还是在Docker上安装，简单方便快捷，如果自己手动安装，不懂点运维基础知识，坑太多了），具体安装教程度娘一堆资料。当前的canal支持MySQL版本包括5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x。
我的MySQL版本是8.0.23，所以canal是支持的。不知道自己安装是什么版本可以通过SELECT VERSION();命令查看。

4.2数据库配置

从上述可知，因为canal是模拟MySQL slave的交互协议，伪装自己为MySQL slave，向MySQL master发送dump协议获取binary log内容对象的，所以需要MySql开启binlog。
●修改mysql.cnf中的配置：

-- 编辑mysql.cnf文件
vim /etc/my.cnf;
-- 在my.cnf上加入如下配置
[mysqld]
log-bin=mysql-bin #开启binlog
binlog-format=ROW #选择ROW模式
server_id=1 #配置MySQL replaction需要定义，不要和canal的slaveId重复
expire-logs-days=10 #binlog日志保留的天数，清除超过10天的日志，防止日志文件过大，导致磁盘空间不足

●授权canal链接MySQL账号具有作为MySQL slave的权限, 如果已有账户可直接grant（我这边是根据官网示例创建一个canal账号来演示）：

-- 先登录MySQL
mysql -u root -p
-- 创建用户，用户名：canal，密码：qwer1234
CREATE USER canal IDENTIFIED BY 'qwer1234';   
-- 授予上的所有权限给canal用户;
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- 刷新权限;
FLUSH PRIVILEGES;

●查看下MySql是否开启binlog日志
是否开启binlog日志：

SHOW VARIABLES LIKE 'log_bin';

 （1）Canal入门-风君雪科技博客
查看binlog日志文件列表：

SHOW BINARY LOGS;

 （1）Canal入门-风君雪科技博客
查看当前正在写入的binlog文件：

SHOW MASTER STATUS;

 （1）Canal入门-风君雪科技博客

4.3Canal配置

安装运行Canal服务端，一定要记得先检查当前Linux系统是否安装了java8环境，如果没有安装启动Canal时候会有如下提示：

[root@dengwu canal]# sh bin/startup.sh
which: no java in (/data/mysql/bin:/data/mysql/lib:/usr/lib64/qt-3.3/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin)
Cannot find a Java JDK. Please set either set JAVA or put java (>=1.5) in your PATH.

 （1）Canal入门-风君雪科技博客
具体安装步骤如下：
●先从Oracle官网下载JDK安装包：
 （1）Canal入门-风君雪科技博客
通过Xftp工具导入到预先创建/app/package安装包目录下，再在/usr目录下创建java目录并解压：

mkdir /usr/java
cd /app/package
tar zxvf jdk-8u291-linux-x64.tar.gz  -C /usr/java

然后配置java环境变量：

vim /etc/profile

用vim编辑器来编辑profile文件，输入i在文件末尾添加以下内容：

export JAVA_HOME=/usr/java/jdk1.8.0_291
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib:$CLASSPATH
export JAVA_PATH=${JAVA_HOME}/bin:${JRE_HOME}/bin
export PATH=$PATH:${JAVA_PATH}

 （1）Canal入门-风君雪科技博客
配置完java环境变量后，:wq保存退出，看看是否生效：

echo $PATH

如果没有生效，让其生效：

source /etc/profile

再瞄瞄java8是否安装成功：

java -version

 （1）Canal入门-风君雪科技博客
●然后下载canal, 访问release页面, 选择需要的包下载, 如最新版本1.1.5为例：
 （1）Canal入门-风君雪科技博客
可以使用wget工具下载：

wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz

或者手动下载，通过Xftp等工具拉入安装包目录（/app/package）中：
 （1）Canal入门-风君雪科技博客
再创建canal安装目录解压安装包：

mkdir /app/canal
cd /app/package
tar zxvf canal.deployer-1.1.5.tar.gz -C /app/canal

然后修改配置：

cd /app/canal
vi conf/example/instance.properties
i
:wq

 （1）Canal入门-风君雪科技博客
●启动canal：

cd /app/canal
sh bin/startup.sh

注：Windows使用startup.bat启动
●查看canal进程是否启动成功：

ps -ef | grep canal

●查看instance的日志：

vi logs/example/example.log

●关闭canal：

sh bin/stop.sh

●在数据库中查看从库信息：

SHOW SLAVE HOSTS;

 （1）Canal入门-风君雪科技博客
查看下canal实例(example)配置是否成功。
●记得把canal端口加入防火墙策略去：

-- 允许通过防火墙
firewall-cmd --permanent --zone=public --add-port=11111/tcp
-- 从防火墙里移除
firewall-cmd --permanent --zone=public --remove-port=11111/tcp
-- 查看端口在防火墙状态
firewall-cmd --permanent --zone=public --query-port=11111/tcp
-- 重启防火墙
firewall-cmd --reload

注：如果是买阿里云服务器，要到阿里云安全组添加允许通过策略。还有Canal Server的默认端口为:11111，若需要修改，可以去/canal/conf目录下的canal.properties配置文件中进行修改。

5.Canal的.NET客户端CanalSharp使用

5.1快速入门

●先安装客户端：

Install-Package CanalSharp

●初始化日志：
CanalSharp使用Microsoft.Extensions.Logging.Abstractions，因为目前主流日志组件，如：nlog、serilog等，全部支持此日志抽象接入，也就是说你可以通过安装nlog、serilog对其的适配，来使用它们，无论是Console App或则是Web App。

var loggerFactory = LoggerFactory.Create(builder =>
{
    builder
        .AddFilter("Microsoft", LogLevel.Debug)
        .AddFilter("System", LogLevel.Information)
        .AddConsole();
});
var logger= loggerFactory.CreateLogger<SimpleCanalConnection>();

●创建连接：

var conn=new SimpleCanalConnection(new SimpleCanalOptions("127.0.0.1",11111,1234),logger);
await conn.ConnectAsync();//连接到Canal Server
await conn.SubscribeAsync();//订阅

●获取数据：

var msg = await conn.GetAsync(1024);

5.2进阶使用

●解析数据
○Entry
上文conn.GetAsync()返回的是一个Entry集合，Entry对应binlog记录，它可能是事务标记也有可能是行数据变化，通过Entry.EntryType来区分，一般事务的标记在业务消费处理时不需要处理。
示例：

var entries = await conn.GetAsync(1024);
foreach (var entry in entries)
{
    //不处理事务标记
    if (entry.EntryType == EntryType.Transactionbegin || entry.EntryType == EntryType.Transactionend)
    {
        continue;
    }
}

Entry.Header包含了一些binlog以及数据库信息：

属性

说明

Entry.Header.LogfileName

binlog文件名

Entry.Header.LogfileOffset

binlog偏移

Entry.Header.SchemaName

mysql schema名称

Entry.Header.TableName

表名

○RowChange
一般在业务处理中，都会需要行数据的变更，将Entry转换为RowChange对象。
示例：

RowChange rowChange = null;
try
{
    rowChange = RowChange.Parser.ParseFrom(entry.StoreValue);
}
catch (Exception e)
{
    _logger.LogError(e);
}

通过RowChange.EventType来Row是什么变化，Update、Delete和Insert对应sql中的update、delete和insert语句，通过RowChange.RowDatas属性，来访问RowChange对象中包含的行变化数据集合。示例，遍历 RowChange 中的行数据：

foreach (var rowData in rowChange.RowDatas)
{
    //删除的数据
    if (eventType == EventType.Delete)
    {
        PrintColumn(rowData.BeforeColumns.ToList());
    }
    //插入的数据
    else if (eventType == EventType.Insert)
    {
        PrintColumn(rowData.AfterColumns.ToList());
    }
    //更新的数据
    else
    {
        _logger.LogInformation("-------> before");
        PrintColumn(rowData.BeforeColumns.ToList());
        _logger.LogInformation("-------> after");
        PrintColumn(rowData.AfterColumns.ToList());
    }
}

private static void PrintColumn(List<Column> columns)
{
    foreach (var column in columns)
    {
        Console.WriteLine($"{column.Name} ： {column.Value}  update=  {column.Updated}");
    }
}

○Column
Column如其名，代表数据库中表的每一列的信息：

属性名

说明

Column.Name

列名

Column.Value

列的值

Column.Updated

列是否被更新

5.3应答机制

应答机制可以保证消费数据的准确性，Canal服务端会记录Client消费的进度，需要客户端发送ACK消息，服务端才会更新进度。类似于在消息队列中的ACK机制，如RabbitMQ。

●自动应答

await conn.GetAsync(1024);//获取数据并自动应答
GetAsync()会在获取数据后，自动向Server发送ack消息。

●手动应答

var msg = await conn.GetWithoutAckAsync(1024);//获取数据
await conn.AckAsync(msg.Id);//手动应答
await conn.RollbackAsync(msg.Id);//回滚

5.4高可用

这里的高可用分为两类，客户端集群和服务端集群。都是采用冷备模式，因为对于binlog数据消费来说，并行处理将会带来数据顺序错乱的问题，当然你可以通过一些复杂的机制去实现，这里不做说明。集群部署需要Zookeeper组件。
●服务端集群
在conf/canal.properties文件中修改zookeeper地址：

canal.zkServers=127.0.0.1:2181

集群中每个实例需配置相同的zookeeper地址。
●客户端集群
客户端集群和服务端集群采用相同的模式，每个实例去抢占锁，获得了锁那么这个实例就运行获取数据，其他实例做冷备。若正在运行消费数据的实例由于网络波动，导致和zookeeper失去连接，那么其他客户端实例不会立即抢占，会等待60s后才执行抢占，给与这个实例恢复的机会。
客户端集群使用的连接对象和快速入门中的不同：ClusterCanalConnection，但使用方法基本相同。
示例：

//初始化日志
var loggerFactory = LoggerFactory.Create(builder =>
            {
                builder
                    .AddFilter("Microsoft", LogLevel.Debug)
                    .AddFilter("System", LogLevel.Information)
                    .AddConsole();
            });
var logger = loggerFactory.CreateLogger<Program>();
//设置zk地址和clientid，统一集群的client必须相同
var conn = new ClusterCanalConnection( new ClusterCanalOptions("localhost:2181", "12350")
//连接到Server                                      loggerFactory);
await conn.ConnectAsync();
//订阅
await conn.SubscribeAsync();
await conn.RollbackAsync(0);
while (true)
{
    try
    {
        //获取数据
        var msg = await conn.GetAsync(1024);
    }
    catch (Exception e)
    {
        _logger.LogError(e,"Error.");
        //发生异常执行重连，此方法只有集群连接对象才有
        await conn.ReConnectAsync();
    }
}

5.5订阅

订阅指过滤表（table）的规则，Canal客户端发送给客户端订阅规则，那么服务端将会推送符合规则的表数据过来，采用正则匹配。
允许所有表：.*\\..*

6.小结

这里这是简单介绍Canal工作原理，能做什么，还有.NET客户端CanalSharp使用，其实Canal涉及知识点还是很多的，例如配置MQ模式、服务集群、Web管理界面部署，多实例等等。后面如果有时间，我还会继续花时间去学习。

参考文献：
CanalSharp文档
CanalSharp