.Net 5 CanalSharp Mysql CDC (增量同步，捕获变更数据)

àì夳堔傛蜴生んèń 2022-08-31 06:58 203阅读 0赞

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

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

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

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

安装服务，可以点击下边的连接。

# [Docker 系列之 Canal (CDC 增量同步,捕获变更数据)][Docker _ Canal _CDC] #

而CanalSharp 就可以直接 Nuget 安装

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

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

以下是代码，也是从github上抄下来的。

github 地址 ：[https://github.com/dotnetcore/CanalSharp][https_github.com_dotnetcore_CanalSharp]

class Program
        {
            private static ILogger _logger;
            static async Task Main(string[] args)
            {
                await SimpleConn();
                //await ClusterConn();
            }
    
            static async Task ClusterConn()
            {
                using var loggerFactory = LoggerFactory.Create(builder =>
                {
                    builder
                        .AddFilter("Microsoft", LogLevel.Debug)
                        .AddFilter("System", LogLevel.Information)
                        .AddConsole();
                });
                _logger = loggerFactory.CreateLogger<Program>();
                var conn = new ClusterCanalConnection(new ClusterCanalOptions("localhost:2181", "12350") { UserName = "canal", Password = "canal" },
                    loggerFactory);
                await conn.ConnectAsync();
                await conn.SubscribeAsync();
                await conn.RollbackAsync(0);
                while (true)
                {
                    try
                    {
                        var msg = await conn.GetAsync(1024);
                        PrintEntry(msg.Entries);
                    }
                    catch (Exception e)
                    {
                        _logger.LogError(e, "Error.");
                        await conn.ReConnectAsync();
                    }
    
                }
            }
    
            static async Task SimpleConn()
            {
                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}");
                }
            }
        }

运行后，结果如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlc3NoZWk_size_16_color_FFFFFF_t_70][]

数据库和表结构如下:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlc3NoZWk_size_16_color_FFFFFF_t_70 1][]

其他的 业务扩展，还得仔细琢磨逻辑。

[https_canalsharp.azurewebsites.net_zh]: https://canalsharp.azurewebsites.net/zh/
[Docker _ Canal _CDC]: https://blog.csdn.net/i2blue/article/details/119036639
[https_github.com_dotnetcore_CanalSharp]: https://github.com/dotnetcore/CanalSharp
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlc3NoZWk_size_16_color_FFFFFF_t_70]: /images/20220829/003b6551c2ca4e568d6e1bbdbeed86f3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlc3NoZWk_size_16_color_FFFFFF_t_70 1]: /images/20220829/0609cf717ae840be87d744b5d45d8249.png