当前位置：首页 > news >正文

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

news 来源：原创 2024/5/19 4:54:42

之前业务需要捕捉到业务数据增量部分，并对其进行宽表处理，这也是其中的一个技术方案，方案主要是用了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}");
        }
    }
}