仓储层到多维数据库

数据模块

1. 从重构的角度,最开始大泥球的架构中,所有的数据都放到一个库中.随着业务发展需要将表进行分组纵向划分,此时一组表就是一个数据模块.
2. 从业务的角度,依据ddd中领域上下文的概念,正好对应一个数据模块.

设计思路

无论是现在流行的微服务,还是以前的SOA,还是DDD中都有模块化思想.模块化也是面向对象松耦合的思想,跟类和类之间关系类似,模块是一组类内聚形成一个组,组和组处理各自的业务.

1. 物理上解决方案中我们把包或者dll看作一个模块.

这类模块主要负责装配(ioc注册,配置加载)等初始化等操作

2. 逻辑上把DDD中的领域上下文和模块模块看作一个概念.
3. DDD通常我们把所有的业务逻辑放到领域层,而领域层中的实体,聚合根等都需要持久化,所以领域中的模块有其特殊的持久化需求.

基于上述分析,在通常说的Module中抽象出一个子概念DataModule,继承自Module.主要负责组织ORM的元数据数据,对应EF的话就是的数据上下文的概念,设计出一个数据模块对应一个数据上下文的概念.而EF的数据上下文对应的就是一个数据库,继而演化成一个数据模块对应一个数据库.从物理上表现就是一个dll对应一个数据库；逻辑上表现为一个数据模块对应一个数据库.

实现步骤

1. 首先在程序启动的时候加载所有模块.
2. 在所有模块中,找到所有的数据模块.
3. 找到数据模块对应的程序集(一个dll).
4. 找到所有模块中继承自Entity的类型.
5. 通常每一个类型即是数据库中一个表.

演进式设计

1. 虽然大家都在说微服务,但是我觉得在公司或者团队成立之初,单进程处理多个业务是很难避免的.单个数据库也在所难免
2. 数据模块设计的初衷就是在大而全的架构体系中,提供逻辑上的隔离,进而如果要实现物理的隔离会相对容易,并且对数据层是透明的,
   只需要修改相应的数据库连接字符串即可.

数据库工厂

数据库连接配置管理

数据的存放的方式和位置最终依赖于数据库的连接,落到代码中就是一个数据库连接字符.
大而全的架构中需要在单个进程中访问多个数据库资源,如果不加以管理势必会混乱.
而最终管理的实际是数据库连接字符串.
数据工厂的目的就是将所有的数据库连接统一的加载,需要使用的时候统一的地方获取.
采用原始的数据库连接配置的方式在应用启动的时候全部加载,这只能满足静态运行的要求.
而不能根据运行时的状态进行动态的分配,所以数据库工厂实际的加载分为两部分;

1. 静态配置直接在启动时进行加载
2. 动态配置按需进行加载并缓存

最终DbFactory的加载时委托给IDbConfigLoader 来完成的,这样姐可以实现的时候加载配置文件,也可以通过运行时根据动态路由按照一定规则生成数据库连接字符串,更加灵活.

数据库连接配置获取

1. 静态配置：前面提到的数据模块,如果将实现步骤进行反向的执行就是获取的方法

根据实体类型定位到数据模块->根据数据模块的名字获取对应的配置.此时的配置可以按照(类型-配置)进行缓存,提高获取效率.

2. 动态配置：首先要定义动态配置的几个要素:数据模块-编号-路由因子-数据库连接字符串.

动态配置是基于静态的,所以获取动态配置的收首先要定位到数据模块.

• 新增数据->获取数据的路由因子->从配置列表中获取(此时生成的id中包含编号这一要素)

• 根据id获取/修改/删除->解析id中的编号->从配置列表获取

• 查询->根据查询条件获取路由因子->从配置列表中获取

• 如果定位不到单条配置则根据有限的条件获取配置集合进行遍历操作

动态配置获取这部分需要结合动态仓储理解

读写分离

结合Repository模式的理解将Repository分离为两类ICommandRepository和IQueryRepository.
但是Repisotory的读写分离并非一个必须选项,所以IRepository继承自ICommandRepository和IQueryRepository.
实际上写部分的逻辑对不同的ORM,不同的数据库技术有所不同 ,所以无论是增删改只能定义接口实现必须关联到具体的技术.
而读却不同.基于Expression的支持,只要对不同的ORM实现类似EntityFramework中LinqProvider的功能即可实现跨ORM和数据库技术的查询.
所以抽象层级如下：

1. ICommandRepository和IQueryRepository
2. IRepository
3. BaseQueryRepisitory（继承自ICommandRepository和IQueryRepository）:只留下一个getall的抽象方法(具体可以参考ABP中repository实现),实现其他所有的查询功能,因为其他功能都可以getall之后处理.这里借助IQueryable延迟加载的特性.
4. BaseRepository(继承自QueryRepisitory和IRepository):实现诸如批量的功能,委派给单个的操作方法.最终需要子类实现的实际只有Add,Modiy,Remove,Getall这四个方法.

这里的实现只是一个思路,具体要集合静态和动态有不同的命名和实现

静态仓储

在前面进行了读写分离和仓储的设计之后这里只需继续对之前的层级继续向下延伸,不过这里因为了区别于动态仓储,这里实际的类名都增加Static.
这里以EF为例实现接口和对象层级如下：

1. IStaticCommandRepository和IStaticQueryRepository
2. IStaticRepository
3. BaseStaticQueryRepisitory(继承同上)
4. BaseStaticRepository(继承同上)
5. StaticQueryRepository(继承自BaseStaticQueryRepisitory)：从UnitOfWork中获取DbSet来实现getall方法
6. StaticCommandRepository(继承自IStaticCommandRepository)：从UnitOfWork中获取DbSet来实现增删改
7. StaticRepository(继承自BaseStaticRepository)：从UnitOfWork中获取DbSet来实现db的模式为 DbMode.Write | DbMode.Read
8. StaticSeparateRepository(继承自StaticRepository)：增删改获取DbSet的时候 DbMode.Write,get获取DbSet的时候是DbMode.Read 用来区分读和写

DbModel 单独读的情况下模式固定为 DbMode.Read 单独写情况下固定为 DbMode.Write,当实现同时存在的时候根据Repository的目的来根据不同的方法来区分.
静态仓储的实现部分跟现在流行的框架并没有区别,最终的区别是在UnitOfWork的注入和创建DbSet背后的逻辑,在后面会进行分析.

动态仓储

在纵向分库的基础上,如果单个库数据量持续增大一样会带来数据过大响应过慢的问题,这时需要对纵向切割库进行横向的切割.具体需要从以下几个点分析

分布式id生成规则

Id生成的要求

1. 无冲突: 多个进程,线程之间生成无冲突
2. 时间线性增长: 随着时间生成的id递增
3. 便捷性:方便使用,如调用方法般简单
4. 速度快 ： 生成速度快,满足每秒的业务要求
5. 数值类型 ：数据库中数值类型比字符串检索要快

传统方式

生成方式	满足	不满足
数据库自增	1,2,5	3,4
Guid	1,3,4	2,5
时间戳	2,3,4,5,	1
统一的服务	1,2,3,5	4

参考：http://www.cnblogs.com/haoxinyue/p/5208136.html

实现

Guid+时间混编 -> 字符串拼接 -> 二进制拼接

将bigint类型的数字转换成64位二进制数据,然后将需要的信息隐藏到id中

具体实现: {AppId:7} + {AppNode:4} + {Time:32} + {Count:14} + {DataNode:6}

理由

1. 通过AppNode解决同一个应用不同进程之间id的冲突
2. 通过Timer解决递增问题
3. 通过Count解决每秒生成id不冲突,保证单进程每秒id的数据
4. 通过DataNode,解决通过id定位到对应数据库的功能

限制

1. 平台最多应用只有128个
2. 单个应用节点只能有16个
3. 单个进程每秒生成的id不超过 16384
4. 单个应用数据库节点不超过64个

动态路由

仓储本身是对数据访问的一个封装.在静态仓储的基础上,一个类型对应到一个纵向切割的模块.
那么横向切割后如何定位到库进行访问就是个难题.

传统的方式

一般都是对id或者某个字段进行hash,但是在读取的时候却需要扫描多个库来获取结果,后续带来的合并,排序等问题会难以解决.

结合仓储

由于利用仓储模式,那么我们假设仓储的每一个方法都可以定位到一个横向切割的库既可以解决传统方式带来的多库扫描的问题.那么我们对仓储的方法进行分类（依据参数,也就是数据）

1. 新增
2. 删除,修改,根据id获取
3. 条件查询

如果对于同一个库的上面三种访问可以建立同样的路由到同一个库即可解决纵向分库的问题.
这种方式我称之为动态路由(IDynamicRouter),依据运行时调用仓储的参数来确定访问的数据库.下面分析三种路由

1. 新增：新增实体继承自IDynamicRouter
2. 删除,修改,根据Id获取: Id生成IDynamicRouter
3. 条件查询: 这里引入IDynamicSpecification(继承自ISpecification和IDynamicRouter);

IDynamicRouter 实际只有一个String属性Coden.
选取实体和Specification中的若干字段根据算法生成一个字符串,然后根据此字符串即可定位到一个数据库.
基于前面的Id生成算法,在插入时候根据路由可以找到唯一的DataNode,当在有id的情况下即可反向定位到一个数据库.

仓储实现(结合EF)

原始的仓储可能见的最多的是 IRepository,但是在动态仓储中,实际需要增加一个抽象维度(抽象维度详见博客的第一篇文章中内容)IRepository<TEntity,ISpecification>
如是抽象层架如下:

1. ICommandRepository 和 IQueryRepository<TEntity, in TSpecification> :这里需要将静态仓储中的TSpecification固定为ISpecification
2. IDynamicCommandRepository 和 IDynamicQueryRepository(因为Command无需条件所有这里抽象维度只有TEntity) ：分别继承自ICommandRepository 和 IQueryRepository
3. IRepository<TEntity, IDynamicSpecification>: ICommandRepository,IQueryRepository<TEntity,TSpecification>
4. IDynamicRepository : IDynamicCommandRepository,
   IDynamicQueryRepository,
   IRepository<TEntity, IDynamicSpecification>
5. BaseDynamicQueryRepository : IDynamicQueryRepository
6. BaseDynamicRepository : BaseDynamicQueryRepository, IDynamicRepository
7. DynamicQueryRepository : BaseDynamicQueryRepository
8. DynamicCommandRepository : IDynamicCommandRepository
9. DynamicRepository : BaseDynamicRepository
10. DynamicSeparateRepository : BaseDynamicRepository

有好几个抽象的维度和层级在里面,所以这里面导致类的层级较多.无论是静态还是动态,base和base以上的都属于框架的内容属于抽象类,以下的都是关联具体技术实现的属于实现类.

仓储综合说明

无论是静态仓储还是动态仓储的层级都较多,主要是集成了读写分离,并且还是可选导致.

最终的使用上都是继承自CommandRepository,QueryRepository,Repository,SeparateRepository.

具体的使用场景是

1. CommandRepository,QueryRepository 只写和只读的场景
2. Repository 单库读写的场景
3. SeparateRepository 同时读写

这里注意3中同时读写的情况,由于无论采用何种技术,写库和读库的同步并不是实时的(实际有几秒的延迟).所以这里代码虽然集成到一起使用方便,但是在使用时要注意避免在一个请求中写完立即读.

工作单元

工作单元的好处在于如何可以对数据库的多个操作一次性提交,对事务比较友好.但是我在设计的时候考虑到静态情况下的事务无论是否读写分离,其实只是对单个库进行操作(读不包含在事务中).而动态情况下却有对多库进行操作的情况(实际在使用中极少出现多库操作).所以分为动态和静态两种,实际上就是单个和多个,只是为了保持之前命名的一致性.

从动态路由的概念中,操作定位到哪个库实际是由请求Repository的参数决定.

如果利用大部分其他架构的IOC注入UnitOfWork到仓储中,此时将会在请求到达Controller决定你的UnitOfWork实例.
而最终数据库即使在简单的静态仓储中都是由到达那个Repository(TEntity的类型可以确定)确定的,所以这里用组合的方式,
仓储的实际注入中只注入DbFactory dbFactory, ContextFactory contextFactory这两个对象.
具体dbfacotry之前已经介绍过,主要管理数据库连接的配置.而contextfacotry这个下文会有说明.

UnitOfWork中有几个关键点.

为了避免分布式事务和重复提交,那么如果一个请求访问多个不同仓储,并且多个仓储的对应的同一个数据库,那么创建出来的DbSet必须是同一个,继而UnitOfWork管理的DbContext也是同一个.此时实际的Context的唯一性和生命周期是由ContextFactory来管理.

实际的工作流程是,

1. 静态仓储：根据实体类型->获取数据模块名字->从DbFacotry拿到数据库配置->传递给UnitOfWork->传递给ContextFactory->获取DbContext
2. 动态仓储：根据实体类型->获取数据某块名字,结合动态路由从Dbfactory拿到数据库配置->传递给UnitOfWork->传递给ContextFactory->获取DbContext

上下文工厂

上下文特指EF的DbContext,根据数据初始化上下文获取DbContext,初始化数据库的上下文定义如下

 public class DbInitContext
    {
        public DbInitContext(DbConfig config, DbModule module, DbMode mode)
        {
            Mode = mode;
            Config = config;
            Module = module;
        }

        public DbMode Mode { get; set; }

        public DbConfig Config { get; set; }

        public DbModule Module { get; set; }

        public string ConnectiongString
        {
            get
            {
                if (Mode == DbMode.Write)
                    return Config.WriteConnectionString;
                if (Mode == DbMode.Read)
                    return Config.ReadConnectionString;
                return Config.NameOrConnectionString;
            }
        }

        public List<Type> Types => Module.EntityTypes;

        public string GetIdentity()
        {
            return $"{Config.StaticCoden} {ConnectiongString}";
        }

        public static Func<Type, bool> IsEntity
        {
            get { return item => item.IsSubclassOf(typeof(Entity)); }
        }
    }

上下文工厂定义如下

using System;
using System.Collections.Concurrent;
using System.Data;
using System.Data.Entity;
using Coralcode.EntityFramework.Extension;
using Coralcode.Framework.Aspect;
using Coralcode.Framework.Data;
using Coralcode.Framework.Exceptions;
using System.Collections.Generic;

namespace Coralcode.EntityFramework.UnitOfWork
{
    [Inject(RegisterType = typeof(ContextFactory), LifetimeManagerType = LifetimeManagerType.PerResolve)]
    public class ContextFactory : IDisposable
    {

        private ConcurrentDictionary<string, CoralDbContext> _contexts = new ConcurrentDictionary<string, CoralDbContext>();
        private List<IDbConnection> connnections = new List<IDbConnection>();

        private static Func<string, DbInitContext, CoralDbContext> _creator;
        private bool _isDispose;

        public ContextFactory()
        {
            if (_creator == null)
                _creator = (item, context) =>
                {
                    var dbContext = new CoralDbContext(context);
                    return dbContext;
                };
        }

        public static void SetContextCreator(Func<string, DbInitContext, CoralDbContext> creator)
        {
            _creator = creator;
        }

        /// <summary>
        /// 创建数据库上下文
        /// </summary>
        /// <param name="context"></param>
        /// <returns></returns>
        public virtual CoralDbContext Create(DbInitContext context)
        {
            if (context == null)
                throw CoralException.ThrowException<DbErrorCode>(item => item.InvalideDbCoden, "上下文为空");
            if (context.Config == null)
                throw CoralException.ThrowException<DbErrorCode>(item => item.InvalideDbCoden, "上下文配置为空");
            if (string.IsNullOrEmpty(context.Config.NameOrConnectionString))
                throw CoralException.ThrowException<DbErrorCode>(item => item.InvalideDbCoden, "连接字符串为空");
            return _contexts.GetOrAdd(context.GetIdentity(), item => _creator(item, context));
        }

        /// <summary>
        /// 获取数据库连接
        /// </summary>
        /// <param name="context"></param>
        /// <returns></returns>
        public virtual IDbConnection GetConnection(DbInitContext context)
        {
            var connection = Database.DEFaultConnectionFactory.CreateConnection(context.ConnectiongString);
            connnections.Add(connection);
            return connection;
        }


        /// <summary>
        /// 获取执行sql的接口
        /// </summary>
        /// <param name="connection"></param>
        /// <returns></returns>
        public virtual ISql GetSqlExetuator(IDbConnection connection)
        {
            return new DapperSql(connection);
        }


        public void DisposeDbContext(string dbIdentity)
        {
            if (string.IsNullOrEmpty(dbIdentity))
                return;
            if (_contexts == null)
                return;
            CoralDbContext context;
            if (_contexts.TryRemove(dbIdentity, out context))
                context.Dispose();
        }

        public void Dispose()
        {
            if (_contexts != null)
            {
                foreach (var context in _contexts)
                {
                    context.Value?.Dispose();
                }
            }
            _contexts?.Clear();
            _contexts = null;
            connnections?.ForEach(item =>
            {
                item.Dispose();
            });
            connnections?.Clear();
            connnections = null;
        }

    }
}

其中SetContextCreator 方法提供可以自定义上下文的扩展,例如Sqlce的,后面再业务组件文章中介绍会提到.

另外dbIdentity 是Dbcontext的一个细节,自带的Dbcontext是根据Dbcontext类型的静态缓存类和表的映射关系.
Dbcontext继承IDbModelCacheKeyProvider之后就可以用dbIdentity来隔离不同数据模块的元数据缓存.
主要是在单体架构中,多库时无需继承框架的Dbcontext即可实现多个上下文元数据管理(具体可查看前面CRUD的数据层设计)

其他意外情况

动态路由意外

大部分情况下数据是可以路由的,但是也免不了不能路由的情况

分页请求

在路由意外的情况中,以分页最难处理,因为分页涉及到排序合并等.

这里我们根据实际情况分析,满足大部分请求快速响应的原则；

1. 分页查询数据量较小,这时全表扫描后内存排序分页成本不高.
2. 数据量较大,此时用户大部分请求会命中前几页和最后几页.

基于以上规则设计如下算法

 /// <summary>
        ///获取分页数据
        /// </summary>
        /// <param name="pageIndex">页码</param>
        /// <param name="pageCount">页大小</param>
        /// <param name="specification">条件</param>
        /// <param name="orderByExpressions">是否排序</param>
        /// <returns>实体的分页数据</returns>
        public PagedList<TEntity> GetPaged(int pageIndex, int pageCount, IDynamicSpecification<TEntity> specification,
            SortExpression<TEntity> orderByExpressions = null)
        {
            if (orderByExpressions == null || !orderByExpressions.IsNeedSort())
                orderByExpressions = new SortExpression<TEntity>(new List<EditableKeyValuePair<Expression<Func<TEntity, dynamic>>, bool>>
                {
                    new EditableKeyValuePair<Expression<Func<TEntity, dynamic>>, bool>(item=>item.Id,false),
                });

            if (pageIndex == 0)
            {
                pageIndex = 1;
            }
            //如果动态路由可用则为单库
            if (!string.IsNullOrEmpty(specification.Coden))
            {
                var set = DynamicGetAll(specification);
                //如果找到了单库
                if (set != null)
                {
                    var queryable = set.Where(specification.SatisfiedBy());
                    int totel = queryable.Count();
                    IEnumerable<TEntity> items = orderByExpressions.BuildSort(queryable).Skip(pageCount * (pageIndex - 1)).Take(pageCount);
                    return new PagedList<TEntity>(totel, pageCount, pageIndex, items.ToList());
                }

            }
           
            //如果找不到单库
            int sum = 0;
            List<IQueryable<TEntity>> entities = new List<IQueryable<TEntity>>();
            foreach (var tmp in DbFactory.GetDynamicDbConfigs(typeof(TEntity)))
            {
                var queryable = DynamicGetAll(new SampleRouter(tmp.DynamicCoden)).Where(specification.SatisfiedBy());
                sum += queryable.Count();
                entities.Add(queryable);
            }
            int newDataIndex = (pageIndex + 1) * pageCount;
            //如果在中值之后则反转排序
            if (sum < pageIndex * pageCount * 2 && pageIndex * pageCount > sum)
            {
                orderByExpressions.Reverse();
                //反转页码
                newDataIndex = sum - pageIndex * pageCount;
                var datas = entities.SelectMany(item => orderByExpressions.BuildSort(item).Take(newDataIndex)).ToList();

                orderByExpressions.Reverse();
                datas = orderByExpressions.BuildSort(datas).Skip(0).Take(pageCount).ToList();
                return new PagedList<TEntity>(sum, pageCount, pageIndex, datas.ToList());
            }
            else
            {
                var datas = entities.SelectMany(item => orderByExpressions.BuildSort(item).Take(newDataIndex))
                    .Skip(pageCount * (pageIndex - 1)).Take(pageCount).ToList();
                return new PagedList<TEntity>(sum, pageCount, pageIndex, datas.ToList());
            }
        }

在这个算法中,中值部分最慢,两端较快:如图

举例:
假设有10个数据库的某个表都存放1w数据.

1. 如果获取第一页数据,只需要从每个表中取10条数据,最后再次合并分页.
2. 如果获取第二页数据,只需要从每个表中取20条数据,最后合并分页.依此类推
3. 如果取最后一页数据, 首先反转排序条件,然后只需要从每个库取最后10条,最后合并分页.

所以最后的性能图如下(!!!!手绘意思下):

总结和展望

动态分库可以归结到数据模型C=f(x),其中C为数据库连接字符串,x为Entity,Specification的字段,甚至是当前请求中应用的某个状态,f为通过X生成Coden的函数

这个典型的应用场景在美团,58等地域性较强的业务中根据省份或者城市分库较为常见.这时候f可能就是一个映射关系(通过key获取value,字典即可).
另外在多租户的情况下,做数据隔离也是比较理想的解决思路.

在更为复杂的情况下C=f(x,y,......),其中C为数据库连接字符串,x,y为Entity或者Specification中某几字段,或者请求的某个状态,f为通过X生成Coden的函数.

另外还可能出现 Cs=f(x,y,......)(少了其中某个参数或几个参数),其中Cs为一组数据库连接字符串

这几个函数需要对业务了解比较清楚,才能实现.

这种分库的方式我总结为多维数据库,其中X,Y,Z等就是不同的维度,每个数据库是多维空间中的点.

通过f定位到一个数据库实际就是多维空间的一个点.而之前的Cs,可能是比多维少一些维度比如三维空间上落在二维平面上的点.

这种思路在阿里mycat中间件,和阿里maxcomputer计算平台的hash分片中都有所体现.

不过相对于来说我的这种实现基于应用程序的改造比较简单,但是通用性会有所不足.

最后: 这些设计也并非一簇而就,在过去两年经过两轮大的重构之后才形成.其中我觉得最重要的是想象力.后面多维数据库的概念更多的是想想的空间.
有兴趣可以留言我们讨论