核心代码分析
最关键的在于获取捕获表信息（系统表中间_CT结尾的数据）。
根据网上资料查取，找到了获取当前捕获表时间区间范围内数据的方式。
见[SQL Server 多表数据增量获取和发布 2.3(https://www.jianshu.com/p/6a400eca6e79)
--10.按照时间范围查询CDC结果
DECLARE @from_lsn BINARY(10),@end_lsn BINARY(10)
DECLARE @start_time DATETIME = '2018-08-01'
DECLARE @end_time DATETIME ='2018-08-30'
SELECT @from_lsn=sys.fn_cdc_map_time_to_lsn('smallest greater than or equal',@start_time)
SELECT @end_lsn=sys.fn_cdc_map_time_to_lsn(' largest less than or equal',@end_time)
SELECT * FROM cdc.fn_cdc_get_all_changes_dbo_Department(@from_lsn,@end_lsn,'all')
数据既然能够通过sql语句获取到，那么逻辑判断就会变得简单，通过分析我们可以发现select * from XXX ,XXX就是上文中讲到的CDC生成的表值函数，表值函数前面相等，可变化的就是架构名_表名称（dbo_Person）
所以我们完全可以通过拼接sql语句得到我们需要的内容，可以默认返回给我们的数据是不友好的，我们还需要自己在做一步设置，将某些字段变成我们好理解的内容
如对下文内容不理解，可翻阅LZ之前的文章
- sys.fn_cdc_map_lsn_to_time(__$start_lsn) AS UpdateTime
- [__$operation] AS Operation
通过查看CDC生成的捕获表我们发现，其实他是在原来的数据表结构上新增了几个字段给我们，其他的表也相同。
那我们在代码中对实体的设计就可以基于继承相同父类的方式，定义一个父类，拥有共同属性
public partial class ExtBase
{
/// <summary>
/// 更新时间
/// </summary>
public DateTime UpdateTime { get; set; }
/// <summary>
/// 操作方式 1 = 删除,2 = 插入,3 = 更新（旧值）,4 = 更新（新值）
/// </summary>
public int Operation { get; set; }
}
其他表都是在自己原来字段的基础上继承当前父类
public class Department : ExtBase
{
public int Id { get; set; }
public string Name { get; set; }
}
public class Person : ExtBase
{
public int Id { get; set; }
public string Name { get; set; }
public int? Age { get; set; }
}
实体类结构完毕后我们开始考虑获取数据的业务逻辑，根据业务我们可以假设获取数据的方法几乎相同，不同的地方就是返回的数据实体集合不同，那我们通过何种方法来完成逻辑的有效封装，这是需要考虑的问题。
经过思考，我构想出了一种方法
1、定义一个抽象基类，在其中定义公共业务逻辑（GetDate）方法，然后定义一个抽象方法，抽象方法需要被子类继承，而子类需要做的就是覆写父类的GetData方法，唯一需要修改的就是传递的实体——可以采用泛型变量的形式去实现
2、等所有的子类构建完成以后，创建一个简单工厂，传递需要的参数，然后根据参数中的唯一标识符，实例化对应的操作类去执行公共方法。
首先是基类抽象类
/// <summary>
/// Cdc 数据捕获服务类
/// </summary>
/// <typeparam name="T"></typeparam>
public abstract class CTBaseService
{
/// <summary>
/// 获取CDC捕获表的数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="schema_table"></param>
/// <param name="startDateTime"></param>
/// <returns></returns>
private List<T> GetRangeList<T>(string schema_table, DateTime startDateTime) where T : class, new()
{
//获取当前需要更新的日期集合列表
var conn = new SqlConnection(StaticConst.Conn);
string query = @"SELECT * ,sys.fn_cdc_map_lsn_to_time(__$start_lsn) AS UpdateTime,[__$operation] AS Operation FROM [cdc].[{2}_CT] WHERE[__$operation] IN(1, 2, 4) AND sys.fn_cdc_map_lsn_to_time(__$start_lsn) > '{0}'
AND sys.fn_cdc_map_lsn_to_time(__$start_lsn) <= '{1}';";
string nowDate = DateTime.Now.ToString();
query = string.Format(query, startDateTime.ToString(), nowDate, schema_table);
var queryList = conn.Query<T>(query).ToList();
return queryList;
}
/// <summary>
/// 抽象方法，由父类实现
/// </summary>
/// <param name="id"></param>
/// <param name="schema_table"></param>
/// <param name="startDateTime"></param>
public abstract void Work(int id, string schema_table, DateTime startDateTime);
/// <summary>
/// 得到CDC捕获数据并插入队列
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="id"></param>
/// <param name="schema_table"></param>
/// <param name="startDateTime"></param>
protected void GetRangeListAndInsertQueue<T>(int id, string schema_table, DateTime startDateTime) where T : ExtBase, new()
{
//获取当前需要更新的日期集合列表
List<T> queryList = GetRangeList<T>(schema_table, startDateTime);
if (queryList.Count > 0)
{
//对集合进行操作 【集合序列化，注意集合反序列化】
string jsonItem = JsonConvert.SerializeObject(queryList);
QueueWorker.Instance.EnqueueItem(jsonItem);
int iret = UpdateServiceLog(id, queryList);
string nowDate = DateTime.Now.ToString();
if (iret > 0)
{
Log4NetHelper.Info(string.Format("表名:{0}\r\n 队列数据：{1} \r\n 插入时间:{2}", schema_table, jsonItem, nowDate));
}
}
}
}
子类实现
我们可以发现子类实现非常好理解，正如上文所说，基类提供了一个抽象接口供子类实现，而子类也真的只需要修改一个实体就可以。如果大家不懂，可以多看几遍来裂解其中的设计方式。
public class DepartmentCT : CTBaseService
{
public override void Work(int id, string schema_table, DateTime startDateTime)
{
base.GetRangeListAndInsertQueue<Department>(id, schema_table, startDateTime);
}
}
public class PersonCT : CTBaseService
{
public override void Work(int id, string schema_table, DateTime startDateTime)
{
base.GetRangeListAndInsertQueue<Person>(id, schema_table, startDateTime);
}
}
最后我们建立一个工厂类，工厂类主要负责接受参数并创建对应的CT帮助类,代码结构如下。根据表名作为唯一标识符字段，创建***CT服务类，然后因为他们继承并覆写了父类抽象方法Work,所以调用.Work方法即可实现获取数据并插入队列的功能。
public class CTCExecuteFactory
{
/// <summary>
/// 执行服务
/// </summary>
/// <param name="id"></param>
/// <param name="schema_table"></param>
/// <param name="updateTime"></param>
public static void ExecuteService(int id, string schemaName, string tableName, DateTime updateTime)
{
CTBaseService service = null;
switch (tableName.ToLower())
{
case ServiceTables.person: service = new PersonCT(); break;
case ServiceTables.department: service = new DepartmentCT(); break;
default: break;
}
if (service != null) service.Work(id, string.Format("{0}_{1}", schemaName, tableName), updateTime);
}
}
其他模块的代码我觉得属于正常理解范围内的东西，不予说明，有兴趣的可自行下载代码查看具体功能。
下载链接
