using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using CanalSharp.Client.Impl;
using CanalSharp.Common.Logging;
using CanalSharp.Logging.NLog;
using Com.Alibaba.Otter.Canal.Protocol;
using NLog;
using NLog.Config;
namespace CanalSharp.SimpleClient
{
class Program
{
static void Main(string[] args)
{
LogManager.Configuration = new XmlLoggingConfiguration("NLog.Config");
var logger = LogManager.GetLogger("test");
logger.Debug("Nlog enabled.");
//设置nlog
CanalSharpLogManager.SetLoggerFactory(new NLogLoggerFactory());
//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(".*\\\\..*");
while (true)
{
//获取数据 1024表示数据大小 单位为字节
var message = connector.Get(1024);
//批次id 可用于回滚
var batchId = message.Id;
if (batchId == -1 || message.Entries.Count <= 0)
{
// Console.WriteLine("=====没有数据了=====");
Thread.Sleep(300);
continue;
}
PrintEntry(message.Entries);
}
}
///
/// 输出数据
///
/// 一个entry表示一个数据库变更
private static void PrintEntry(List 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());
}
}
}
}
}
///
/// 输出每个列的详细数据
///
///
private static void PrintColumn(List columns)
{
foreach (var column in columns)
{
//输出列明 列值 是否变更
Console.WriteLine($"{column.Name} : {column.Value} update= {column.Updated}");
}
}
}
}