mysql数据库变更监控(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 =

#position info，需要改成自己的数据库信息

canal.instance.master.address = 127.0.0.1:

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-

#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", , destination, "", "");

            //连接 Canal

            connector.Connect();

            //订阅，同时传入 Filter。Filter是一种过滤规则，通过该规则的表数据变更才会传递过来

            //允许所有数据 .*\\..*

            //允许某个库数据 库名\\..*

            //允许某些表 库名.表名,库名.表名

            connector.Subscribe(".*\\..*");

            while (true)

            {

                //获取数据 1024表示数据大小 单位为字节

                var message = connector.Get();

                //批次id 可用于回滚

                var batchId = message.Id;

                if (batchId == - || message.Entries.Count <= )

                {

                    Thread.Sleep();

                    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}");

            }

        }

    }

运行代码，去到数据库中改一下某行数据：

可以看到我们代码收集到变更信息：

本篇就介绍到这里了, 至于如何将变更同步到es，那是属于es操作的范畴，可参考 https://www.elastic.co/guide/en/elasticsearch/client/net-api/current/elasticsearch-net.html

以上内容源于：

https://github.com/alibaba/canal/wiki/QuickStart

https://github.com/dotnetcore/CanalSharp

巴特西

mysql数据库变更监控(canal)

mysql主备复制实现

canal的工作原理：

最新文章

热门文章