SQL Server 多表数据增量获取和发布 4

--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')

    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; }
    }

    /// <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);
        }
    }

    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);
        }
    }