.NET 6 Mysql Canal (CDC 增量同步,捕获变更数据) 案例版

Posted 2022-09-05 dotNET跨平台

之前业务需要捕捉到业务数据增量部分，并对其进行宽表处理，这也是其中的一个技术方案，方案主要是用了CDC的技术。

CDC 全称是 Change Data Capture，捕获变更数据，是一个比较广泛的概念，只要是能够捕获所有数据的变化，比如数据库捕获完整的变更日志记录增、删、改等，都可以称为 CDC。该功能被广泛应用于数据同步、更新缓存、微服务间同步数据等场景。

而 Canal 是阿里巴巴旗下的一个 CDC中间件，使其成为mysql的从机来接收实时的数据修改事件。

整体大致架构：Mysql -> Canal Server(Docker) -> C# BLL (Canal Client)

整体大概是这样的架构。

确定Mysql是否开启BinLog

如果没有开启，是捕获不到日志的。

show variables like 'log_bin';  -- binlog 开关
show variables like 'binlog_format';  -- binlog格式 我这边是row

这样就可以下一步了。

Canal服务安装

先安装Docker

docker pull canal/canal-server:latest

大致启动命令如下:

命令如下
Usage:
  run.sh [CONFIG]
example 1 :
  run.sh -e canal.instance.master.address=127.0.0.1:3306 \\
         -e canal.instance.dbUsername=canal \\
         -e canal.instance.dbPassword=canal \\
         -e canal.instance.connectionCharset=UTF-8 \\
         -e canal.instance.tsdb.enable=true \\
         -e canal.instance.gtidon=false \\
         -e canal.instance.filter.regex=.*\\\\\\..*
example 2 :
  run.sh -e canal.admin.manager=127.0.0.1:8089 \\
         -e canal.admin.port=11110 \\
         -e canal.admin.user=admin \\
         -e canal.admin.passwd=4ACFE3202A5FF5CF467898FC58AAB1D615029441

我这边自己处理后的命令

docker run -d -it -h --name=canal-server ^
-p 11110:11110 -p 11111:11111 -p 11112:11112 -p 9100:9100 -m 4096m ^
-e canal.instance.master.address=192.168.1.8:3306 ^
-e canal.instance.dbUsername=root ^
-e canal.instance.dbPassword=123456 ^
-e canal.instance.connectionCharset=UTF-8 ^
-e canal.instance.tsdb.enable=true ^
-e canal.instance.gtidon=false ^
-e canal.instance.filter.regex=.*\\\\\\..* ^
canal/canal-server

Windows使用Docker Desktop出现exit 139错误

docker linux 内核7以下会出现此类问题

建议直接升级内核或者 采用 如下配置

创建或修改 C:\\Users(用户名).wslconfig，里面写入

[wsl2]
kernelCommandLine = vsyscall=emulate

然后，重启电脑主机即可。

CanalSharp （canal client）客户端

CanalSharp 阿里云的解决方案，需要两部分

Canal 服务端要和Mysql 连在一起（目前我是用docker部署的服务）

另外一部分就是 CanalSharp 单独的客户端服务(.Net 6 服务)

CanalSharp文档 可以参考：https://canalsharp.azurewebsites.net/zh/

而CanalSharp 就可以直接 Nuget 安装

另外一个包是日志，不需要的话，可以不需要，但是，我这个Demo还是需要的

Install-Package CanalSharp -Version 1.1.0
Install-Package Microsoft.Extensions.Logging.Console -Version 6.0.0-preview.6.21352.12

具体的示例

一个简单的数据库表结构，作为演示用。

CREATE TABLE `NewTable` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `name` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

业务代码

class Program

    private static ILogger _logger;
    static async Task Main(string[] args)
    
        await SimpleClusterConn();
        Console.WriteLine("同步结束!");
        Console.ReadLine();
    

    static async Task SimpleClusterConn()
    
        using var loggerFactory = LoggerFactory.Create(builder =>
        
            builder
                .AddFilter("Microsoft", LogLevel.Debug)
                .AddFilter("System", LogLevel.Information)
                .AddConsole();
        );
        _logger = loggerFactory.CreateLogger<Program>();
        var conn = new SimpleCanalConnection(new SimpleCanalOptions("127.0.0.1", 11111, "12349"), loggerFactory.CreateLogger<SimpleCanalConnection>());
        await conn.ConnectAsync();
        await conn.SubscribeAsync();
        await conn.RollbackAsync(0);
        while (true)
        
            var msg = await conn.GetAsync(1024);
            PrintEntry(msg.Entries);
            await Task.Delay(300);
        
    

    private static void PrintEntry(List<Entry> entries)
    
        foreach (var entry in entries)
        
            if (entry.EntryType == EntryType.Transactionbegin || entry.EntryType == EntryType.Transactionend)
            
                continue;
            

            RowChange rowChange = null;

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

            if (rowChange != null)
            
                EventType eventType = rowChange.EventType;

                _logger.LogInformation(
                    $"================> binlog[entry.Header.LogfileName:entry.Header.LogfileOffset] , name[entry.Header.SchemaName,entry.Header.TableName] , eventType :eventType");

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

启动后，如下图所示：

提示订阅成功。

数据库表操作

1. 1. 插入

1. 1. 更新

1. 1. 删除

至此，插入，更新，删除的信息都有了。可以根据这些信息，对业务表进行监控等行为来提升业务功能。

总结

秋高气爽，终于不用那么热了。

CDC功能总得来说是不可缺的，特别是跟大数据相关的业务，这个步骤总是不可少，无非是别人做和自己来做这件事情，而CDC做的好的，也就阿里了。

用着有问题，就上git上问问，实在不行改改源码。

阅

一键三连呦！，感谢大佬的支持，您的支持就是我的动力!

代码地址

https://github.com/kesshei/CanalSharpDemo.git

https://gitee.com/kesshei/CanalSharpDemo.git