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

核心代码分析

最关键的在于获取捕获表信息(系统表中间_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)

image.png

所以我们完全可以通过拼接sql语句得到我们需要的内容,可以默认返回给我们的数据是不友好的,我们还需要自己在做一步设置,将某些字段变成我们好理解的内容 如对下文内容不理解,可翻阅LZ之前的文章

  • sys.fn_cdc_map_lsn_to_time(__$start_lsn) AS UpdateTime
  • [__$operation] AS Operation

通过查看CDC生成的捕获表我们发现,其实他是在原来的数据表结构上新增了几个字段给我们,其他的表也相同。

image.png

image.png

那我们在代码中对实体的设计就可以基于继承相同父类的方式,定义一个父类,拥有共同属性

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

其他模块的代码我觉得属于正常理解范围内的东西,不予说明,有兴趣的可自行下载代码查看具体功能。 下载链接

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Web项目聚集地

手写一个Mybatis框架

在手写自己的Mybatis框架之前,我们先来了解一下Mybatis,它的源码中使用了大量的设计模式,阅读源码并观察设计模式在其中的应用,才能够更深入的理解源...

1092
来自专栏木宛城主

SharePoint CAML In Action——Part II

在SharePoint中,相对于Linq to SharePoint而言,CAML是轻量化的。当然缺点也是显而易见的,"Hard Code"有时会让你抓狂。在实...

1995
来自专栏spring源码深度学习

java基础io流——File的告白(重温经典)

创建成功返回true,如果存在就不创建返回false,创建一个文件时需要确保当前文件夹存在,所有要异常处理。

1693
来自专栏JackieZheng

Hadoop阅读笔记(六)——洞悉Hadoop序列化机制Writable

  酒,是个好东西,前提要适量。今天参加了公司的年会,主题就是吃、喝、吹,除了那些天生话唠外,大部分人需要加点酒来作催化剂,让一个平时沉默寡言的码农也能成为一个...

2345
来自专栏java架构师

Stream篇(1)

最近在看一个写的很好的博客,为了加深记忆,把自认为重要的东西,一边看,一边记在这里 一、什么是流、字节序列、字节 一条河中有一条鱼游过,这条鱼就是一个字节,这个...

3037
来自专栏MasiMaro 的技术博文

windows 纤程

纤程本质上也是线程,是多任务系统的一部分,纤程为一个线程准并行方式调用多个不同函数提供了一种可能,它本身可以作为一种轻量级的线程使用。它与线程在本质上没有区别,...

1712
来自专栏专注 Java 基础分享

Hibernate框架学习之注解映射实体类

     前面的相关文章中,我们已经介绍了使用XML配置文件映射实体类及其各种类型的属性的相关知识。然而不论是时代的潮流还是臃肿繁杂的配置代码告诉我们,注解配置...

2219
来自专栏我叫刘半仙

原自己手写一个Mybatis框架(简化)

       继上一篇手写SpringMVC之后,我最近趁热打铁,研究了一下Mybatis。MyBatis框架的核心功能其实不难,无非就是动态代理和jdbc的操...

2K6
来自专栏Hadoop数据仓库

HAWQ技术解析(十) —— 过程语言

        HAWQ支持用户自定义函数(user-defined functions,UDF),还支持给HAWQ内部的函数起别名。编写UDF的语言可以是SQ...

4765
来自专栏chenssy

【死磕Sharding-jdbc】---group by结果合并(2)

在sharding-jdbc源码之group by结果合并(1)中主要分析了sharding-jdbc如何在GroupByStreamResultSetMerg...

1052

扫码关注云+社区

领取腾讯云代金券