Canal数据库变更监控

Posted 2021-02-17 wikiz

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Canal数据库变更监控相关的知识，希望对你有一定的参考价值。

背景：

1. 一些项目的基础功能会有Audit Trace, 以记录系统用户所做过的所有记录。

2. 实时备份数据，比如mysql主从复制，一个用于面向应用，一个用于对应用数据库的实时备份。

3. 实时收集关系型数据库变更，将数据保存在nosql数据库中，以提供快速检索，一个较为实用的场景就是实现地将mysql数据变更同步到elastic search 或者 mongo db。

下面，将介绍如何通过canal，将mysql 数据变更同步到elastic search 。

首先我们了解一下什么是canal?

mysql主备复制实现

从上层来看，复制分成三步：

master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events，可以通过show binlog events进行查看）；
slave将master的binary log events拷贝到它的中继日志(relay log)；
slave重做中继日志中的事件，将改变反映它自己的数据。

canal的工作原理：

原理相对比较简单：

canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议
mysql master收到dump请求，开始推送binary log给slave(也就是canal)
canal解析binary log对象(原始为byte流)

安装步骤：

访问：https://github.com/alibaba/canal/releases ，会列出所有历史的发布版本包下载方式，比如以1.0.17版本为例子：

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

mkdir /tmp/canal
tar zxvf canal.deployer-1.0.17.tar.gz  -C /tmp/canal

配置修改

vi conf/example/instance.properties

#################################################
## mysql serverId
canal.instance.mysql.slaveId = 1234
#position info，需要改成自己的数据库信息
canal.instance.master.address = 127.0.0.1:3306
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.timestamp =


#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =


#username/password，需要改成自己的数据库信息
canal.instance.dbUsername = canal

canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8


#table regex
canal.instance.filter.regex = ...


#################################################

准备启动

sh bin/startup.sh

查看日志

vi logs/canal/canal.log

关闭

sh bin/stop.sh

下面试下在代码中，获取到mysql变更：

首先安装下 canal 客户端 nuget包

Install-Package CanalSharp.Client

static void Main(string[] args)
        {
            //canal 配置的 destination，默认为 example
            var destination = "example";
            //创建一个简单 CanalClient 连接对象（此对象不支持集群）传入参数分别为 canal 地址、端口、destination、用户名、密码
            var connector = CanalConnectors.NewSingleConnector("192.168.1.23", 11111, destination, "", "");
            //连接 Canal
            connector.Connect();
            //订阅，同时传入 Filter。Filter是一种过滤规则，通过该规则的表数据变更才会传递过来
            //允许所有数据 .*\\..*
            //允许某个库数据 库名\\..*
            //允许某些表 库名.表名,库名.表名
            connector.Subscribe(".*\\..*");
            while (true)
            {
                //获取数据 1024表示数据大小 单位为字节
                var message = connector.Get(1024);
                //批次id 可用于回滚
                var batchId = message.Id;
                if (batchId == -1 || message.Entries.Count <= 0)
                {
                    Thread.Sleep(300);
                    continue;
                }

                PrintEntry(message.Entries);
            }
        }

        /// <summary>
        /// 输出数据
        /// </summary>
        /// <param name="entrys">一个entry表示一个数据库变更</param>
        private static void PrintEntry(List<Entry> entrys)
        {
            foreach (var entry in entrys)
            {
                if (entry.EntryType == EntryType.Transactionbegin || entry.EntryType == EntryType.Transactionend)
                {
                    continue;
                }

                RowChange rowChange = null;

                try
                {
                    //获取行变更
                    rowChange = RowChange.Parser.ParseFrom(entry.StoreValue);
                }
                catch (Exception e)
                {
                    Console.WriteLine(e.Message);
                }

                if (rowChange != null)
                {

                    //by the changed entry‘s table name and record id. get the changed order(full info with any children records) form mysql and save it to es.

                    //to do it, boys !


                    //变更类型 insert/update/delete 等等
                    EventType eventType = rowChange.EventType;
                    //输出binlog信息 表名 数据库名 变更类型
                    Console.WriteLine(
                        $"================> binlog[{entry.Header.LogfileName}:{entry.Header.LogfileOffset}] , name[{entry.Header.SchemaName},{entry.Header.TableName}] , eventType :{eventType}");

                    //输出 insert/update/delete 变更类型列数据
                    foreach (var rowData in rowChange.RowDatas)
                    {
                        if (eventType == EventType.Delete)
                        {
                            PrintColumn(rowData.BeforeColumns.ToList());
                        }
                        else if (eventType == EventType.Insert)
                        {
                            PrintColumn(rowData.AfterColumns.ToList());
                        }
                        else
                        {
                            Console.WriteLine("-------> before");
                            PrintColumn(rowData.BeforeColumns.ToList());
                            Console.WriteLine("-------> after");
                            PrintColumn(rowData.AfterColumns.ToList());
                        }
                    }
                }
            }
        }

        /// <summary>
        /// 输出每个列的详细数据
        /// </summary>
        /// <param name="columns"></param>
        private static void PrintColumn(List<Column> columns)
        {
            foreach (var column in columns)
            {
                //输出列明 列值 是否变更
                Console.WriteLine($"{column.Name} ： {column.Value}  update=  {column.Updated}");
            }
        }
    }