开发者社区> 晓晨Master> 正文

CanalSharp-mysql数据库binlog的增量订阅&消费组件Canal的.NET客户端

简介:
+关注继续查看

一.前言

CanalSharp是阿里巴巴开源项目mysql数据库binlog的增量订阅&消费组件 Canal 的.NET客户端,关于什么是 Canal?又能做什么?我会在后文为大家一一介绍。CanalSharp 这个项目,是由我和 WithLin (主要贡献) 完成,并将一直进行维护的Canal的.NET客户端项目。目前开源在github:https://github.com/CanalSharp/CanalSharp/ 希望大家多多支持,旨在为.NET开发者提供一个友好的对接Canal的选择,为.NET社区生态做贡献。

二.Canal介绍

1.背景

早期,阿里巴巴B2B公司因为存在杭州和美国双机房部署,存在跨机房同步的业务需求。不过早期的数据库同步业务,主要是基于trigger的方式获取增量变更,不过从2010年开始,阿里系公司开始逐步的尝试基于数据库的日志解析,获取增量变更进行同步,由此衍生出了增量订阅&消费的业务,从此开启了一段新纪元。

  1. 目前内部版本已经支持mysql和oracle部分版本的日志解析,当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)

基于日志增量订阅&消费支持的业务:

  1. 数据库镜像
  2. 数据库实时备份
  3. 多级索引 (卖家和买家各自分库索引)
  4. search build
  5. 业务cache刷新
  6. 价格变化等重要业务消息

2.工作原理

2.1 mysql主备复制实现

1537856325274

从上层来看,复制分成三步:

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

2.2 Canal的工作原理

1537856417016

原理相对比较简单:

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

以上内容摘自Canal项目官方资料 https://github.com/alibaba/canal

3.Canal的安装以及使用

Canal的安装以及使用请查阅官方文档,本文不在赘述。 https://github.com/alibaba/canal/wiki

三.CanalSharp介绍

1.工作原理

CanalSharp 是 Canal 的 .NET 客户端,它与 Canal 是采用的Socket来进行通信的,传输协议是TCP,交互协议采用的是 Google Protocol Buffer 3.0。

2.工作流程

1.Canal连接到mysql数据库,模拟slave

2.CanalSharp与Canal建立连接

2.数据库发生变更写入到binlog

5.Canal向数据库发送dump请求,获取binlog并解析

4.CanalSharp向Canal请求数据库变更

4.Canal发送解析后的数据给CanalSharp

5.CanalSharp收到数据,消费成功,发送回执。(可选)

6.Canal记录消费位置。

以一张图来表示:

1537860226808

3.应用场景

CanalSharp作为Canal的客户端,其应用场景就是Canal的应用场景。关于应用场景在Canal介绍一节已有概述。这里我举一些实际的使用例子:

1.代替使用轮询数据库方式来监控数据库变更,有效改善轮询耗费数据库资源。

2.根据数据库的变更实时更新搜索引擎,比如电商场景下商品信息发生变更,实时同步到商品搜索引擎 Elasticsearch、solr等

3.根据数据库的变更实时更新缓存,比如电商场景下商品价格、库存发生变更实时同步到redis

4.数据库异地备份、数据同步

5.根据数据库变更触发某种业务,比如电商场景下,创建订单超过xx时间未支付被自动取消,我们获取到这条订单数据的状态变更即可向用户推送消息。

6.将数据库变更整理成自己的数据格式发送到kafka等消息队列,供消息队列的消费者进行消费。

四.CanalSharp的使用

1.使用前的准备

使用 CanalSharp 之前,必然要先准备好mysql数据库以及Canal才行,这个步骤请直接查阅Canal官方文档 https://github.com/alibaba/canal/wiki 。但是为了让大家能快速跑通CanalSharp,CanalSharp 项目为大家提供了一个通过 docker-compose 同时运行 mysql和canal。

2.通过docker-compose运行mysql和canal:

git clone https://github.com/CanalSharp/CanalSharp.git
cd docker
docker-compose up -d

出现下图表示运行成功:

1537866674285

3.使用navicat等数据库管理工具连接mysql

ip:运行docker的服务器ip

mysql用户:root

mysql密码:000000

mysql端口:4406

默认提供了一个test数据库,然后有一张名为test的表。

1537866852816

4.创建一个 .NET Core 控制台项目

5.添加 Nuget 程序包

Install-Package CanalSharp.Client

1537867132107

6.编码

也可以直接下载源码运行 Sample 项目 https://github.com/CanalSharp/CanalSharp/tree/master/sample/CanalSharp.SimpleClient

(1)建立连接

   //canal 配置的 destination,默认为 example
    var destination = "example";
    //创建一个简单CanalClient连接对象(此对象不支持集群)传入参数分别为 canal地址、端口、destination、用户名、密码
    var connector = CanalConnectors.NewSingleConnector("127.0.0.1", 11111, destination, "", "");
    //连接 Canal
    connector.Connect();
    //订阅,同时传入Filter,如果不传则以Canal的Filter为准。Filter是一种过滤规则,通过该规则的表数据变更才会传递过来
    connector.Subscribe(".*\\\\..*");

(2)获取数据

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

(3)输出数据

/// <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)
        {

        }

        if (rowChange != null)
        {
            //变更类型 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}");
    }
}

7.测试运行

首次运行会输出一堆数据,那些都是初始化运行创建表的数据,忽略即可

运行项目,然后一次执行sql观察输出:

insert into test values(1000,'111');
update test set name='222' where id=1000;
delete from test where id=1000;

通过新标签页打开图片查看大图

可以看见我们分别执行 insert、update、delete 语句,我们的CanalSharp都获取到了数据库变更。

五.使用Canal的经验

1.mysql数据库版本有要求:5.7.13, 5.6.10,、5.5.18和5.1.40/48,不一定非要满足小版本号的要求,比如 5.7.x、5.6.x、5.5.x都应该可以,但是实际需要自己做测试。前面的具体版本号是Canal官方提供的资料,但是博主公司用的mysql 的版本是5.5.60,是可以正常使用Canal的。

2.mysql数据binlog的格式强烈建议设置为row

3.Canal并非必须连接到master数据库,它同样可以连接到slave数据库,只是从库出了需要开启写入binlog以外还需要设置 log-slave-updates 开启。

4.如果生产环境已经存在mysql集群,且集群主库的binlog格式为mixed,mysql数据库集群的主库binlog格式可以不用改依然为 mixed,设置某一个从库binlog格式配置为 row,让Canal连接从库,这样可以避免对生产环境的mysql集群产生影响。

5.mysql支持Statement,MiXED,以及ROW三种格式的binlog为什么推荐使用row格式binlog,经过博主实际测试,使用row格式兼容性是最好的,实际可以自己测试。

六.结束语

CanalSharp的介绍到这里就结束了,如果觉得这个项目有用的欢迎大家来个 star 。后续将会写几篇文章介绍更详细的使用方法以及实战。

七.资料

CanalSharp 开源地址:https://github.com/CanalSharp/CanalSharp

Canal 开源地址:https://github.com/alibaba/canal

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
使用Asp.net MVC 2.0 +.NET 4.0 出现 “从客户端 ... 中检测到有潜在危险的 Request.Form 值”错误的解决办法
我们在用Asp.net 开发 Web Form页面时,通常要提交包含Html内容的数据给后台程序代码处理时,会为页面设置 ValidateRequest="false" 属性,设置后,讨厌的 "从客户端 ... 中检测到有潜在危险的 Request.Form 值”的错误乖乖消失了。
648 0
一句代码实“.NET技术”现批量数据绑定[上篇]
  对于一个以数据处理为主的应用中的UI层,我们往往需要编写相当多的代码去实现数据绑定。如果界面上的控件和作为数据源的实体类型之间存储某种约定的映射关系,我们就可以实现批量的数据绑定。为了验证这种想法,我写了一个小小的组件。
632 0
一句代“.NET技术”码实现批量数据绑定[上篇]
  对于一个以数据处理为主的应用中的UI层,我们往往需要编写相当多的代码去实现数据绑定。如果界面上的控件和作为数据源的实体类型之间存储某种约定的映射关系,我们就可以实现批量的数据绑定。为了验证这种想法,我写了一个小小的组件。
636 0
一句“.NET技术”代码实现批量数据绑定[下篇]
  《上篇》主要介绍如何通过DataBinder实现批量的数据绑定,以及如何解决常见的数据绑定问题,比如数据的格式化。接下来,我们主要来谈谈DataBinder的设计,看看它是如何做到将作为数据源实体的属性值绑定到界面对应的控件上的。
621 0
Asp.Net Web API 2第三课——.NET客户端调用Web API
原文:Asp.Net Web API 2第三课——.NET客户端调用Web API Asp.Net Web API 导航     Asp.Net Web API第一课——入门http://www.cnblogs.com/aehyok/p/3432158.html       Asp.Net Web API第二课——CRUD操作http://www.cnblogs.com/aehyok/p/3434578.html 前言 本教程演示从一个控制台应用程序,使用HttpClient调用Web API。
1131 0
一句代码实现批量数据绑定“.NET技术”[下篇]
  《上篇》主要介绍如何通过DataBinder实现批量的数据绑定,以及如何解决常见的数据绑定问题,比如数据的格式化。接下来,我们主要来谈谈DataBinder的设计,看看它是如何做到将作为数据源实体的属性值绑定到界面对应的控件上的。
621 0
阿里云物联网 .NET Core 客户端 | CZGL.AliIoTClient:1. 连接阿里云物联网
1) 阿里云物联网 阿里云物联网支持多种通讯协议,CZGL.AliIoTClient 使用了 MQTT 通讯协议,通讯库为 M2MQTT 。阿里云物联网数据传输有两种数据传输方式,分别是 透传 和 Alink json,两种方式只在属性读/写、事件上报、服务调用这四种 Topic 上有差异, 其它连接通讯、普通 Topic、响应等,无差别。
1265 0
mysqlbinlog同步数据
mysqlbinlog同步数据
30 0
【.Net MF网络开发板研究-05】Socket编程之客户端
官方示例中也包括了SocketClient,不过其完成的功能一是DNS解析,二是网站数据接收。我们需要大幅度改进一下,让SocketClient程序和PC上的Socket Server进行通信。
466 0
+关注
晓晨Master
微软最有价值专家(Microsoft MVP)、开源爱好者、.NET 技术爱好者、.NET Core Community (NCC) PMC。喜欢关注和研究前沿技术,热衷于技术和经验分享,拥有长期的技术博客写作经验,同时活跃于开源社区,热爱开源。
18
文章
3
问答
文章排行榜
最热
最新
相关电子书
更多
OceanBase 入门到实战教程
立即下载
阿里云图数据库GDB,加速开启“图智”未来.ppt
立即下载
实时数仓Hologres技术实战一本通2.0版(下)
立即下载