倉儲層到多維數據庫

時間 2019-12-14

標籤倉儲多維數據庫欄目 SQL 简体版

原文原文鏈接

數據模塊

從重構的角度,最開始大泥球的架構中,全部的數據都放到一個庫中.隨着業務發展須要將表進行分組縱向劃分,此時一組表就是一個數據模塊.
從業務的角度,依據ddd中領域上下文的概念,正好對應一個數據模塊.

設計思路

不管是如今流行的微服務,仍是之前的SOA,仍是DDD中都有模塊化思想.模塊化也是面向對象鬆耦合的思想,跟類和類之間關係相似,模塊是一組類內聚造成一個組,組和組處理各自的業務.html

物理上解決方案中咱們把包或者dll看做一個模塊.算法

這類模塊主要負責裝配(ioc註冊,配置加載)等初始化等操做sql
邏輯上把DDD中的領域上下文和模塊模塊看做一個概念.
DDD一般咱們把全部的業務邏輯放到領域層,而領域層中的實體,聚合根等都須要持久化,因此領域中的模塊有其特殊的持久化需求.數據庫

基於上述分析,在一般說的Module中抽象出一個子概念DataModule,繼承自Module.主要負責組織ORM的元數據數據,對應EF的話就是的數據上下文的概念,設計出一個數據模塊對應一個數據上下文的概念.而EF的數據上下文對應的就是一個數據庫,繼而演化成一個數據模塊對應一個數據庫.從物理上表現就是一個dll對應一個數據庫；邏輯上表現爲一個數據模塊對應一個數據庫.c#

實現步驟

首先在程序啓動的時候加載全部模塊.
在全部模塊中,找到全部的數據模塊.
找到數據模塊對應的程序集(一個dll).
找到全部模塊中繼承自Entity的類型.
一般每個類型便是數據庫中一個表.

演進式設計

雖然你們都在說微服務,可是我以爲在公司或者團隊成立之初,單進程處理多個業務是很難避免的.單個數據庫也在所不免
數據模塊設計的初衷就是在大而全的架構體系中,提供邏輯上的隔離,進而若是要實現物理的隔離會相對容易,而且對數據層是透明的,
只須要修改相應的數據庫鏈接字符串便可.

數據庫工廠

數據庫鏈接配置管理

數據的存放的方式和位置最終依賴於數據庫的鏈接,落到代碼中就是一個數據庫鏈接字符.
大而全的架構中須要在單個進程中訪問多個數據庫資源,若是不加以管理勢必會混亂.
而最終管理的實際是數據庫鏈接字符串.
數據工廠的目的就是將全部的數據庫鏈接統一的加載,須要使用的時候統一的地方獲取.
採用原始的數據庫鏈接配置的方式在應用啓動的時候所有加載,這隻能知足靜態運行的要求.
而不能根據運行時的狀態進行動態的分配,因此數據庫工廠實際的加載分爲兩部分;緩存

靜態配置直接在啓動時進行加載
動態配置按需進行加載並緩存

最終DbFactory的加載時委託給IDbConfigLoader 來完成的,這樣姐能夠實現的時候加載配置文件,也能夠經過運行時根據動態路由按照必定規則生成數據庫鏈接字符串,更加靈活.架構

數據庫鏈接配置獲取

靜態配置：前面提到的數據模塊,若是將實現步驟進行反向的執行就是獲取的方法

根據實體類型定位到數據模塊->根據數據模塊的名字獲取對應的配置.此時的配置能夠按照(類型-配置)進行緩存,提升獲取效率.app

動態配置：首先要定義動態配置的幾個要素:數據模塊-編號-路由因子-數據庫鏈接字符串.框架

動態配置是基於靜態的,因此獲取動態配置的收首先要定位到數據模塊.分佈式

新增數據->獲取數據的路由因子->從配置列表中獲取(此時生成的id中包含編號這一要素)
根據id獲取/修改/刪除->解析id中的編號->從配置列表獲取
查詢->根據查詢條件獲取路由因子->從配置列表中獲取
若是定位不到單條配置則根據有限的條件獲取配置集合進行遍歷操做

動態配置獲取這部分須要結合動態倉儲理解

讀寫分離

結合Repository模式的理解將Repository分離爲兩類ICommandRepository和IQueryRepository.
可是Repisotory的讀寫分離並不是一個必須選項,因此IRepository繼承自ICommandRepository和IQueryRepository.
實際上寫部分的邏輯對不一樣的ORM,不一樣的數據庫技術有所不一樣 ,因此不管是增刪改只能定義接口實現必須關聯到具體的技術.
而讀卻不一樣.基於Expression的支持,只要對不一樣的ORM實現相似EntityFramework中LinqProvider的功能便可實現跨ORM和數據庫技術的查詢.
因此抽象層級以下：

ICommandRepository和IQueryRepository
IRepository
BaseQueryRepisitory（繼承自ICommandRepository和IQueryRepository）:只留下一個getall的抽象方法(具體能夠參考ABP中repository實現),實現其餘全部的查詢功能,由於其餘功能均可以getall以後處理.這裏藉助IQueryable延遲加載的特性.
BaseRepository(繼承自QueryRepisitory和IRepository):實現諸如批量的功能,委派給單個的操做方法.最終須要子類實現的實際只有Add,Modiy,Remove,Getall這四個方法.

這裏的實現只是一個思路,具體要集合靜態和動態有不一樣的命名和實現

靜態倉儲

在前面進行了讀寫分離和倉儲的設計以後這裏只需繼續對以前的層級繼續向下延伸,不過這裏由於了區別於動態倉儲,這裏實際的類名都增長Static.
這裏以EF爲例實現接口和對象層級以下：

IStaticCommandRepository和IStaticQueryRepository
IStaticRepository
BaseStaticQueryRepisitory(繼承同上)
BaseStaticRepository(繼承同上)
StaticQueryRepository(繼承自BaseStaticQueryRepisitory)：從UnitOfWork中獲取DbSet來實現getall方法
StaticCommandRepository(繼承自IStaticCommandRepository)：從UnitOfWork中獲取DbSet來實現增刪改
StaticRepository(繼承自BaseStaticRepository)：從UnitOfWork中獲取DbSet來實現db的模式爲 DbMode.Write | DbMode.Read
StaticSeparateRepository(繼承自StaticRepository)：增刪改獲取DbSet的時候 DbMode.Write,get獲取DbSet的時候是DbMode.Read 用來區分讀和寫

DbModel 單獨讀的狀況下模式固定爲 DbMode.Read 單獨寫狀況下固定爲 DbMode.Write,當實現同時存在的時候根據Repository的目的來根據不一樣的方法來區分.
靜態倉儲的實現部分跟如今流行的框架並無區別,最終的區別是在UnitOfWork的注入和建立DbSet背後的邏輯,在後面會進行分析.

動態倉儲

在縱向分庫的基礎上,若是單個庫數據量持續增大同樣會帶來數據過大響應過慢的問題,這時須要對縱向切割庫進行橫向的切割.具體須要從如下幾個點分析

分佈式id生成規則

Id生成的要求

無衝突: 多個進程,線程之間生成無衝突
時間線性增加: 隨着時間生成的id遞增
便捷性:方便使用,如調用方法般簡單
速度快：生成速度快,知足每秒的業務要求
數值類型：數據庫中數值類型比字符串檢索要快

傳統方式

生成方式	知足	不知足
數據庫自增	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}

理由

經過AppNode解決同一個應用不一樣進程之間id的衝突
經過Timer解決遞增問題
經過Count解決每秒生成id不衝突,保證單進程每秒id的數據
經過DataNode,解決經過id定位到對應數據庫的功能

限制

平臺最多應用只有128個
單個應用節點只能有16個
單個進程每秒生成的id不超過 16384
單個應用數據庫節點不超過64個

動態路由

倉儲自己是對數據訪問的一個封裝.在靜態倉儲的基礎上,一個類型對應到一個縱向切割的模塊.
那麼橫向切割後如何定位到庫進行訪問就是個難題.

傳統的方式

通常都是對id或者某個字段進行hash,可是在讀取的時候卻須要掃描多個庫來獲取結果,後續帶來的合併,排序等問題會難以解決.

結合倉儲

因爲利用倉儲模式,那麼咱們假設倉儲的每個方法均可以定位到一個橫向切割的庫既能夠解決傳統方式帶來的多庫掃描的問題.那麼咱們對倉儲的方法進行分類（依據參數,也就是數據）

新增
刪除,修改,根據id獲取
條件查詢

若是對於同一個庫的上面三種訪問能夠創建一樣的路由到同一個庫便可解決縱向分庫的問題.
這種方式我稱之爲動態路由(IDynamicRouter),依據運行時調用倉儲的參數來肯定訪問的數據庫.下面分析三種路由

新增：新增實體繼承自IDynamicRouter
刪除,修改,根據Id獲取: Id生成IDynamicRouter
條件查詢: 這裏引入IDynamicSpecification(繼承自ISpecification和IDynamicRouter);

IDynamicRouter 實際只有一個String屬性Coden.
選取實體和Specification中的若干字段根據算法生成一個字符串,而後根據此字符串便可定位到一個數據庫.
基於前面的Id生成算法,在插入時候根據路由能夠找到惟一的DataNode,當在有id的狀況下便可反向定位到一個數據庫.

倉儲實現(結合EF)

原始的倉儲可能見的最多的是 IRepository ,可是在動態倉儲中,實際須要增長一個抽象維度(抽象維度詳見博客的第一篇文章中內容)IRepository<TEntity,ISpecification>
如是抽象層架以下:

ICommandRepository 和 IQueryRepository<TEntity, in TSpecification> :這裏須要將靜態倉儲中的TSpecification固定爲ISpecification
IDynamicCommandRepository 和 IDynamicQueryRepository(由於Command無需條件全部這裏抽象維度只有TEntity) ：分別繼承自ICommandRepository 和 IQueryRepository
IRepository<TEntity, IDynamicSpecification >: ICommandRepository ,IQueryRepository<TEntity,TSpecification>
IDynamicRepository : IDynamicCommandRepository ,
IDynamicQueryRepository ,
IRepository<TEntity, IDynamicSpecification >
BaseDynamicQueryRepository : IDynamicQueryRepository
BaseDynamicRepository : BaseDynamicQueryRepository , IDynamicRepository
DynamicQueryRepository : BaseDynamicQueryRepository
DynamicCommandRepository : IDynamicCommandRepository
DynamicRepository : BaseDynamicRepository
DynamicSeparateRepository : BaseDynamicRepository

有好幾個抽象的維度和層級在裏面,因此這裏面致使類的層級較多.不管是靜態仍是動態,base和base以上的都屬於框架的內容屬於抽象類,如下的都是關聯具體技術實現的屬於實現類.

倉儲綜合說明

不管是靜態倉儲仍是動態倉儲的層級都較多,主要是集成了讀寫分離,而且仍是可選致使.

最終的使用上都是繼承自CommandRepository,QueryRepository,Repository,SeparateRepository.

具體的使用場景是

CommandRepository,QueryRepository 只寫和只讀的場景
Repository 單庫讀寫的場景
SeparateRepository 同時讀寫

這裏注意3中同時讀寫的狀況,因爲不管採用何種技術,寫庫和讀庫的同步並非實時的(實際有幾秒的延遲).因此這裏代碼雖然集成到一塊兒使用方便,可是在使用時要注意避免在一個請求中寫完當即讀.

工做單元

工做單元的好處在於如何能夠對數據庫的多個操做一次性提交,對事務比較友好.可是我在設計的時候考慮到靜態狀況下的事務不管是否讀寫分離,其實只是對單個庫進行操做(讀不包含在事務中).而動態狀況下卻有對多庫進行操做的狀況(實際在使用中極少出現多庫操做).因此分爲動態和靜態兩種,實際上就是單個和多個,只是爲了保持以前命名的一致性.

從動態路由的概念中,操做定位到哪一個庫實際是由請求Repository的參數決定.

若是利用大部分其餘架構的IOC注入UnitOfWork到倉儲中,此時將會在請求到達Controller決定你的UnitOfWork實例.
而最終數據庫即便在簡單的靜態倉儲中都是由到達那個Repository(TEntity的類型能夠肯定)肯定的,因此這裏用組合的方式,
倉儲的實際注入中只注入DbFactory dbFactory, ContextFactory contextFactory這兩個對象.
具體dbfacotry以前已經介紹過,主要管理數據庫鏈接的配置.而contextfacotry這個下文會有說明.

UnitOfWork中有幾個關鍵點.

爲了不分佈式事務和重複提交,那麼若是一個請求訪問多個不一樣倉儲,而且多個倉儲的對應的同一個數據庫,那麼建立出來的DbSet必須是同一個,繼而UnitOfWork管理的DbContext也是同一個.此時實際的Context的惟一性和生命週期是由ContextFactory來管理.

實際的工做流程是,

靜態倉儲：根據實體類型->獲取數據模塊名字->從DbFacotry拿到數據庫配置->傳遞給UnitOfWork->傳遞給ContextFactory->獲取DbContext
動態倉儲：根據實體類型->獲取數據某塊名字,結合動態路由從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的數據層設計)

其餘意外狀況

動態路由意外

大部分狀況下數據是能夠路由的,可是也免不了不能路由的狀況

分頁請求

在路由意外的狀況中,以分頁最難處理,由於分頁涉及到排序合併等.

這裏咱們根據實際狀況分析,知足大部分請求快速響應的原則；

分頁查詢數據量較小,這時全表掃描後內存排序分頁成本不高.
數據量較大,此時用戶大部分請求會命中前幾頁和最後幾頁.

基於以上規則設計以下算法

/// <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數據.