实现效果:
设计思路:
1. 开启数据库及表的cdc,定时查询cdc表数据,封装sql语句(通过执行类型,主键;修改类型的cdc数据只取最后更新的记录),添加到离线数据表;
2. 线程定时查询离线数据表,更新远程库数据;
3. 远程库数据被更改又会产生cdc数据,对此数据进行拦截;
配置文件说明:
{
"AsyncInterval": 30000,
"Drivers": [
{
"RefreshTime": 5000,
"Enable": 1,
"SrcConnect": "Data Source=192.168.8.77;Initial Catalog=master;User ID=sa;Pwd=Nflg1234",
"SrcMap": [ "dbsync2|student,table1,table2,table3", "dbsync3|*" ],
"SrcUpdateCDC": 1,
"DstConnect": [ "Data Source=192.168.8.81;Initial Catalog=master;User ID=sa;Pwd=Nflg1234" ]
}
]
}
或
{
"AsyncInterval": 25000,
"Drivers": [
{
"RefreshTime": 10000,
"Enable": 1,
"SrcConnect": "Data Source=192.168.8.77;Initial Catalog=master;User ID=sa;Pwd=Nflg1234",
"SrcMap": [ "testsync1|*"],
"SrcUpdateCDC": 1,
"DstConnect": [ "Data Source=192.168.8.81;Initial Catalog=master;User ID=sa;Pwd=Nflg1234" ]
},
{
"RefreshTime": 10000,
"Enable": 1,
"SrcConnect": "Data Source=192.168.8.81;Initial Catalog=master;User ID=sa;Pwd=Nflg1234",
"SrcMap": [ "testsync1|*" ],
"SrcUpdateCDC": 1,
"DstConnect": [ "Data Source=192.168.8.77;Initial Catalog=master;User ID=sa;Pwd=Nflg1234" ]
}
]
}
1. 设置同步间隔时间
2. 根据不同的配置文件,加载不同的模式,多驱动(Drivers 1主1备-单向同步,1主1主-双向同步,2主1备-多库汇总),多机同步(DstConnect),多库同多表同步(SrcMap,dbsync2|*表示监听该数据库下的所有表),设置刷新时间(RefreshTime),是否启用(Enable),是否重置cdc数据(SrcUpdateCDC)
数据表说明:
async_data 离线数据表
id 主键自增 INTEGER
connect_str 连接字符串 NVARCHAR(255)
excute_sql 需要同步的sql语句 NVARCHAR(255)
cdc_time cdc时间 DATETIME
event_time event时间 DATETIME
db_name 数据库名 NVARCHAR(255)
table_name 表名 NVARCHAR(255)
table_pk 表主键 NVARCHAR(255)
excute_type 执行类型(I/U/D) NVARCHAR(255)
sqlserver cdc表(日志表)中如果一条id多次更新,取最新一条数据
sqlite asy_data表(离线数据表),入库时,查dbname + table + pk,无记录则添加,有记录比较cdc记录时间,如果时间更新则更新sql语句
特殊数据处理:
uniqueidentifier类型的数据转为NULL,数据中含有'的替换''
核心代码:
using SqlServerAsync.Util.config; using SqlServerAsync.Util.sqlite; using SqlServerAsync.Util.sqlite.model; using System; using System.Collections.Generic; using System.Data; using System.Threading; namespace SqlServerAsync.Util { public class SqlServerCDC { public void Listen(Driver driver) { var update_cdc = driver.SrcUpdateCDC == 1 ? true : false; var enable = driver.Enable == 1 ? true : false; foreach (var map in driver.SrcMap) { StartCDC(driver.SrcConnect, driver.DstConnect, driver.RefreshTime, enable, update_cdc, map); } } void StartCDC(string srcconnect, List<string> dstconnect, int refreshTime, bool enable, bool update_cdc, string map) { try { var freeSql = new FreeSql.FreeSqlBuilder() .UseConnectionString(FreeSql.DataType.SqlServer, srcconnect) .UseNoneCommandParameter(true)// 不使用参数化 .UseAutoSyncStructure(false)// 不同步表结构 .Build(); var arrayMap = map.Split('|'); var db = arrayMap[0]; var tbs = arrayMap[1]; string[] arrayTB = null; var dstStr = string.Join("#", dstconnect); if (!enable) { Program.AddLog($"禁用监听,来源={srcconnect},目标数={dstconnect.Count},目标={dstStr},db={db},Tables={tbs}"); return; } string sql = string.Empty; Dictionary<string, Table> dicTable = new Dictionary<string, Table>(); Program.AddLog($"启用监听,来源={srcconnect},目标数={dstconnect.Count},目标={dstStr},db={db},Tables={tbs}"); if ("*" == tbs) { // 查询db下所有表名 sql = $"use {db};select TABLE_NAME from {db}.information_schema.tables where TABLE_SCHEMA='dbo' and TABLE_NAME not in('systranschemas','sysdiagrams')"; DataTable dtAll = freeSql.Ado.ExecuteDataTable(sql); var rowCount = dtAll.Rows.Count; if (rowCount > 0) arrayTB = new string[rowCount]; for (int i = 0; i < rowCount; i++) { arrayTB[i] = dtAll.Rows[i]["TABLE_NAME"].ToString(); } } else { arrayTB = tbs.Split(','); } if (null == arrayTB || 0 == arrayTB.Length) { Program.AddLog($"数据库{db},查无数据表 ×"); return; } // 开启SQL Server数据库CDC sql = $"use {db};if exists(select 1 from {db}.sys.databases where name='{db}' and is_cdc_enabled=0)\n" + "begin\n" + $"exec {db}.sys.sp_cdc_enable_db\n" + "end"; freeSql.Ado.ExecuteNonQuery(sql); // 查询库cdc是否开启成功 sql = $"use {db};select is_cdc_enabled from {db}.sys.databases where name='{db}'"; DataTable dtCDC_DB = freeSql.Ado.ExecuteDataTable(sql); if (dtCDC_DB.Rows.Count <= 0 || !Convert.ToBoolean(dtCDC_DB.Rows[0]["is_cdc_enabled"])) { Program.AddLog($"数据库CDC开启失败({db}) ×"); return; } Program.AddLog($"数据库CDC开启成功({db}) √"); foreach (var table in arrayTB) { if (string.IsNullOrEmpty(table)) continue; if (update_cdc) { // 关闭单张表的CDC功能 sql = $"use {db};if exists(select 1 from {db}.sys.tables where name='{table}' AND is_tracked_by_cdc=1)\n" + "begin\n" + $"exec {db}.sys.sp_cdc_disable_table @source_schema='dbo',@source_name='{table}',@capture_instance='dbo_{table}'" + "end"; freeSql.Ado.ExecuteNonQuery(sql); } // 开启单张表的CDC功能 sql = $"use {db};if exists(select 1 from {db}.sys.tables where name='{table}' AND is_tracked_by_cdc=0)\n" + "begin\n" + $"exec {db}.sys.sp_cdc_enable_table\n" + "@source_schema='dbo',\n" + $"@source_name='{table}',\n" + "@capture_instance=NULL,\n" + "@supports_net_changes=1,\n" + "@role_name=NULL\n" + "end"; freeSql.Ado.ExecuteNonQuery(sql); // 查询表cdc是否开启成功 sql = $"use {db};select is_tracked_by_cdc from {db}.sys.tables WHERE name='{table}'"; DataTable dtCDC_TB = freeSql.Ado.ExecuteDataTable(sql); if (dtCDC_TB.Rows.Count <= 0 || !Convert.ToBoolean(dtCDC_TB.Rows[0]["is_tracked_by_cdc"])) { Program.AddLog($"数据表CDC开启失败({table}) ×"); continue; } Program.AddLog($"数据表CDC开启成功({table}) √"); Table tb = new Table() { Name = table }; // 获取字段名,是否主键,字段类型 sql = $"use {db};SELECT distinct col.name AS 'Name', idx.is_primary_key as 'IsPK',TYPE_NAME(system_type_id) as 'Type'\n" + $"FROM sys.columns col\n" + $"LEFT JOIN sys.index_columns idxcol ON col.object_id=idxcol.object_id AND col.column_id=idxcol.column_id\n" + $"LEFT JOIN sys.indexes idx ON idxcol.object_id=idx.object_id AND idxcol.index_id=idx.index_id\n" + $"WHERE col.object_id=OBJECT_ID('{table}')"; List<Field> lstField = freeSql.Ado.Query<Field>(sql); foreach (var field in lstField) { var ispk = Convert.ToBoolean(field.IsPK); if (ispk) { tb.LstPKField.Add(field);// 主键,用于更新删除 } else { tb.LstDataField.Add(field); } } dicTable.Add(table, tb); } Program.AddLog($"监听成功,{db}"); // 定时轮询 ThreadPool.QueueUserWorkItem(delegate { Dictionary<string, string> dicTBUpdatePK = new Dictionary<string, string>(); while (true) { try { foreach (var item in dicTable) { dicTBUpdatePK.Clear(); var table_name = item.Key; var tableEntity = item.Value; // cdc表查询 //__$start_lsn :与相应更改的提交事务关联的日志序列号(LSN) //__$end_lsn : (在 SQL Server 2008中,此列始终为 NULL) //__$seqval :对事务内的行更改顺序 //__$operation :源表DML操作 var cdc_table_name = $"{db}.cdc.dbo_{table_name}_CT"; sql = $"use {db};select sys.fn_cdc_map_lsn_to_time(__$start_lsn) as cdctime,* from {cdc_table_name}";// 查询cdc时间 var dt = freeSql.Ado.ExecuteDataTable(sql); table_name = $"{db}.dbo." + table_name; for (int i = 0; i < dt.Rows.Count; i++) { var row = dt.Rows[i]; var lstPKField = tableEntity.LstPKField; var lstDataField = tableEntity.LstDataField; var cdctime = Convert.ToDateTime(row["cdctime"]); var operation = Convert.ToInt32(row["__$operation"]); var seqval = (byte[])(row["__$seqval"]);// __$start_lsn代表事件时间,并发时,会有相同的情况,改用__$seqval var str_seqval = BitConverter.ToString(seqval, 0).Replace("-", string.Empty); if (3 == operation) { continue; } var sql_cdc_execute = string.Empty; string table_pk = string.Empty; foreach (var field1 in lstPKField) { table_pk += field1.Name + "='" + row[field1.Name] + "' and "; } table_pk = table_pk.Substring(0, table_pk.Length - 5); string cdc_dic_pk = table_name + ";" + table_pk; // cdc表中过滤多条表中一条记录多次更新,取最新一条数据(查询过的数据利用字典存储) string str_seqval1 = string.Empty; if (4 == operation) { if (dicTBUpdatePK.ContainsKey(cdc_dic_pk)) { str_seqval1 = dicTBUpdatePK[cdc_dic_pk]; } else { // 查询多次更新后的最新值 sql = $"use {db};select top 1 __$seqval from {cdc_table_name} where {table_pk} and __$operation=4 order by __$seqval desc"; var dtlsn = freeSql.Ado.ExecuteDataTable(sql); var seqval1 = (byte[])(dtlsn.Rows[0]["__$seqval"]); str_seqval1 = BitConverter.ToString(seqval1, 0).Replace("-", string.Empty); dicTBUpdatePK.Add(cdc_dic_pk, str_seqval1); } } // 删除cdc表数据 sql = $"use {db};delete from {cdc_table_name} where __$seqval=CONVERT(BINARY(10), '{str_seqval}', 2)"; freeSql.Ado.ExecuteNonQuery(sql); string excute_type = string.Empty; switch (operation) { case 1: // 删除 excute_type = BaseEnum.Delete; sql_cdc_execute = $"delete from {table_name} where {table_pk}"; break; case 2: // 插入 excute_type = BaseEnum.Insert; string insertField = string.Empty; string insertValue = string.Empty; foreach (var field1 in lstPKField) { insertField += field1.Name + ","; insertValue += HandleSpecialData(field1.Type, row[field1.Name]) + ","; } foreach (var field2 in lstDataField) { insertField += field2.Name + ","; insertValue += HandleSpecialData(field2.Type, row[field2.Name]) + ","; } insertField = insertField.Substring(0, insertField.Length - 1); insertValue = insertValue.Substring(0, insertValue.Length - 1); sql_cdc_execute = $"insert into {table_name} ({insertField}) values({insertValue})"; break; case 3: break; case 4: // 修改 if (str_seqval == str_seqval1)// 最新的数据 { excute_type = BaseEnum.Update; string updateData = string.Empty; foreach (var field2 in lstDataField) { updateData += field2.Name + "=" + HandleSpecialData(field2.Type, row[field2.Name]) + ","; } updateData = updateData.Substring(0, updateData.Length - 1); sql_cdc_execute = $"update {table_name} set {updateData} where {table_pk}"; } break; } if (!string.IsNullOrEmpty(sql_cdc_execute)) { foreach (var dst in dstconnect) { bool add = true; string key1 = srcconnect + "_" + table_name + "_" + table_pk; // A同步B,B更新后,CDC日志返回A,这边做截取 if (Program.DicExecuted.ContainsKey(key1)) { add = false; string removedValue; Program.DicExecuted.TryRemove(key1, out removedValue); } else { // 修改以最后时间的数据为准 var entity = SqliteHelper.GetUpdateAsyncData(db, table_name, table_pk); if (null == entity) { var asyncdata = new AsyncData() { ConnectStr = dst, ExcuteSQL = sql_cdc_execute, CDCTime = cdctime, EventTime = DateTime.Now, DBName = db, TableName = table_name, TablePK = table_pk, ExcuteType = excute_type }; SqliteHelper.InsertAsyncData(asyncdata); } else { // 比较时间 if (DateTime.Compare(entity.CDCTime, cdctime) < 0) { SqliteHelper.UpdateAsyncData(dst, sql_cdc_execute, entity.Id); } else { add = false; } } if (add) { if (dst.Contains("192.168.8.81")) { Console.WriteLine("111"); } Program.AddLog($"添加,dst:{dst},sql:{sql_cdc_execute}"); } } } } } } } catch (Exception ex) { Program.AddLog($"Listen Error,ex:{ex.Message}"); } Thread.Sleep(refreshTime); } }); } catch (Exception ex) { Program.AddLog($"[Error] 初始化CDC异常,errmsg:{ex.Message}"); } } /// <summary> /// 特殊数据类型处理 1. uniqueidentifier为空时,设置为NULL;2. 单引号,转成双号 /// </summary> /// <param name="val"></param> /// <returns></returns> public string HandleSpecialData(string type, object val) { if (null == val) return string.Empty; string ret = val.ToString(); bool special = false; if ("uniqueidentifier" == type.ToLower())// 特殊数据类型处理 { if (string.IsNullOrEmpty(ret)) { special = true; ret = "NULL"; } } if (!special) { if (ret.Contains("'")) { ret = ret.Replace("'", "''");// 把单引号转成双引号 } ret = $"'{ret}'"; } return ret; } } public class Table { public string Name { get; set; } public List<Field> LstPKField { get; set; } = new List<Field>(); public List<Field> LstDataField { get; set; } = new List<Field>(); } public class Field { public string Name { get; set; } public string IsPK { get; set; } public string Type { get; set; }// GUID,uniqueidentifier为空时,改为NULL } }