首页 > 其他分享 >Dapper NetCore 分区实战

Dapper NetCore 分区实战

时间：2023-11-17 15:35:15浏览次数：57  
标签：string NetCore 分区 tableName id sql var Dapper public

在上一篇中介绍了基于Dapper的NetCore分表，本篇旨在介绍基于Dapper的NetCore分区，废话不多说开搞吧！

模拟业务场景：基于公司所在地区对表建立分区

设计公司表结构,其中TableAttribute标识表名，PartitionAttribute标识当前表是分区结构，Property代表按照某个属性分区

public abstract class Entity<TKey>
{
    [ExplicitKey]
    public virtual TKey Id { get; set; }
}

[Table("Company")]
[Partition(property:"Region")]
public class Company : Entity<Guid>
{
    public string Name { get; set; }
    public string Region { get; set; }
}
Company 
[AttributeUsage(AttributeTargets.Property)]
public class ExplicitKey : Attribute
{
}

public class PartitionAttribute : Attribute
{
    public string Property { get; set; }
    public PartitionAttribute(string property)
    {
        this.Property = property;
    }
}
Attribute

有了表结构之后设计Repository,首先是BaseRepository除了基本的CRUD，还定义一个虚方法 protected virtual Func<string, string>? CreateScriptFunc { get; set; }要求子类根据特定的需求创建表及定义一个Schema设计的初衷是做数据隔离或者安全隔离。

using Dapper.Contrib.Extensions;
using System.Data;
using System.Reflection;

namespace API.Dapper.Repository.V2
{
    public abstract class BaseRepositoryV2<T> where T : class, new()
    {
        protected IDbConnectionSchema DbConnectionSchema { get; set; }
        protected string Schema { get => DbConnectionSchema.Schema; }
        protected IDbTransaction Transaction { get; private set; }
        protected IDbConnection DbConnection => DbConnectionSchema.DbConnection;
        protected Int32 CommandTimeout { get; set; } = 30;
        protected static string _tableName => typeof(T).GetCustomAttribute<TableAttribute>()?.Name ?? typeof(T).Name;
        protected static IEnumerable<string> _properties => typeof(T).GetProperties().Select(p => p.Name);

        protected BaseRepositoryV2(IDbConnectionSchema dbConnectionSchema, IDbTransaction dbTransaction) 
        {
            DbConnectionSchema = dbConnectionSchema;
            Transaction = dbTransaction;
        }

        protected virtual Func<string, string>? CreateScriptFunc { get; set; }

        public async Task<Int32> AddAsync(IEnumerable<T> entities, string tableName = default)
        {
            var sql = string.Format(Sql.Insert.BuildSchema(Schema), tableName ?? _tableName, string.Join(",", _properties), string.Join(",", _properties.Select(p => $"@{p}")));
            return await DbConnection.ExecuteAsync(sql, entities, Transaction, CommandTimeout, CommandType.Text);
        }

        public async Task<IEnumerable<T>> GetAllAsync(string tableName = default)
        {
            var sql = string.Format(Sql.Select.BuildSchema(Schema), string.Join(",", _properties), tableName ?? _tableName, string.Empty);
            return await DbConnection.QueryAsync<T>(sql, transaction: Transaction, commandTimeout: CommandTimeout, commandType: CommandType.Text);
        }

        public async Task<T> GetAsync(Guid id, string tableName = default)
        {
            var sql = string.Format(Sql.Select.BuildSchema(Schema), string.Join(",", _properties), tableName ?? _tableName, $"WHERE Id = @Id");
            var param = new DynamicParameters();
            param.Add("Id", id);
            return await DbConnection.QuerySingleOrDefaultAsync<T>(sql, param, transaction: Transaction, commandTimeout: CommandTimeout, commandType: CommandType.Text);
        }

        public async Task<IEnumerable<T>> GetAsync(IEnumerable<string> selectedProperties, IEnumerable<string> whereProperties, string tableName = default)
        {
            var sql = string.Format(Sql.Select.BuildSchema(Schema), string.Join(",", selectedProperties), tableName ?? _tableName, string.Join(" AND ", whereProperties.Select(p => $"{p}=@{p}")));
            return await DbConnection.QueryAsync<T>(sql, transaction: Transaction, commandTimeout: CommandTimeout, commandType: CommandType.Text);
        }

        public async Task<IEnumerable<T>> GetAsync(IEnumerable<string> selectedProperties, string whereClause, DynamicParameters sqlParams, string tableName = default)
        {
            var sql = string.Format(Sql.Select.BuildSchema(Schema), string.Join(",", selectedProperties), tableName ?? _tableName, whereClause);
            return await DbConnection.QueryAsync<T>(sql, sqlParams, transaction: Transaction, commandTimeout: CommandTimeout, commandType: CommandType.Text);
        }

        public async Task<Int32> UpdateAsync(IEnumerable<T> entities, IEnumerable<string> updateProperties, IEnumerable<string> whereProperties, string tableName = default)
        {
            var updateClause = string.Join(",", _properties.Intersect(updateProperties).Select(p => $"{p}=@{p}"));
            var whereClause = $"WHERE {string.Join(" AND ", _properties.Intersect(whereProperties).Select(p => $"{p}=@{p}"))}";
            var sql = string.Format(Sql.Update.BuildSchema(Schema), tableName ?? _tableName, updateClause, whereClause);
            return await DbConnection.ExecuteAsync(sql, entities, transaction: Transaction, commandTimeout: CommandTimeout, commandType: CommandType.Text);
        }

        public async Task<Int32> DeleteAsync(IEnumerable<Guid> ids, string tableName = default)
        {
            var sql = string.Format(Sql.Delete.BuildSchema(Schema), tableName ?? _tableName, $"WHERE Id IN @Ids");
            var param = new DynamicParameters();
            param.Add("Ids", ids);
            return await DbConnection.ExecuteAsync(sql, param, transaction: Transaction, commandTimeout: CommandTimeout, commandType: CommandType.Text);
        }
    }
}
BaseRepository

BaseRepository中基础的CURD不能满足分区的业务需求，还需要一些特定的逻辑，比如写入数据的时候需要判断是否有分区Function，分区Schema及是否已经创建分区Range values等，所以定义了PartionRepository继承自BaseRepository完成以上操作

using API.Dapper.Attributes;
using API.Dapper.Repository.V2;
using Dapper;
using System.Data;
using System.Reflection;

namespace API.Dapper.Repository;

public abstract class PartitionRepository<T> : BaseRepositoryV2<T> where T : class, new()
{
    public PartitionRepository(IDbConnectionSchema dbConnectionSchema, IDbTransaction dbTransaction)
        : base(dbConnectionSchema, dbTransaction)
    {
    }

    public async Task<Int32> AddAsync(IEnumerable<T> entities)
    {
        if (typeof(T).GetCustomAttribute<PartitionAttribute>() != null)
        {
            if (!await DbConnection.IsTableExist(Schema, _tableName, Transaction, CommandTimeout))
            {
                await DbConnection.ExecuteAsync(this.CreateScriptFunc?.Invoke(_tableName), transaction: Transaction, commandTimeout: CommandTimeout, commandType: CommandType.Text);
                var partitionValues = GetPartitionValues(entities);
                var existPartitionValues = await DbConnection.GetPartitionValues(_tableName, Schema, transaction: Transaction, commandTimeout: CommandTimeout);
                if (partitionValues.Except(existPartitionValues).Any())
                    await DbConnection.CreatePartitionValues(partitionValues.Except(existPartitionValues), _tableName, Schema, transaction: Transaction, commandTimeout: CommandTimeout);
            }
        }
        return await base.AddAsync(entities);
    }

    private IEnumerable<string> GetPartitionValues(IEnumerable<T> entities)
    {
        var partitionAttribute = typeof(T).GetCustomAttribute(typeof(PartitionAttribute)) as PartitionAttribute;
        var propertyInfo = typeof(T).GetProperties().Where(p => p.Name.Equals(partitionAttribute!.Property)).FirstOrDefault();
        if (propertyInfo != null)
        {
            return entities.Select(e => propertyInfo.GetValue(e, null)).Select(v => $"{v}");
        }
        return default;
    }
}
PartitionRepository

接下来定义CompanyRepository继承自PartitionRepository完成业务Repository

using API.Dapper.Entity;
using System.Data;


namespace API.Dapper.Repository.V2
{
    public class CompanyRepositoryV2 : PartitionRepository<Company>
    {
        private static string _defaultPartitionValue => "CHN";
        public CompanyRepositoryV2(IDbConnectionSchema dbConnectionSchema, IDbTransaction dbTransaction) 
            : base(dbConnectionSchema, dbTransaction)
        {
        }

        protected override Func<string, string> CreateScriptFunc => tableName =>
           _tableName switch
           {
               APITable.Company =>
                $"CREATE PARTITION FUNCTION [{Schema}_{tableName}PartitionFun](nvarchar(256)) AS RANGE RIGHT FOR VALUES ('{_defaultPartitionValue}');" +
                $"CREATE PARTITION SCHEME [{Schema}_{tableName}PartitonSchema] AS PARTITION [{Schema}_{tableName}PartitionFun] ALL TO ([PRIMARY]);" +
                @"CREATE TABLE [{schema}].".BuildSchema(Schema) + $@"[{tableName}](
                [Id] [uniqueidentifier] NOT NULL,
                [Name] [nvarchar](256) NOT NULL,
                [Region] [nvarchar](256) NOT NULL)
                ON [{Schema}_{tableName}PartitonSchema]([Region]);"+
                "CREATE CLUSTERED INDEX [Clustered_Region] ON [{schema}].".BuildSchema(Schema) + $"[{tableName}]([Region] ASC);"+
                "CREATE NONCLUSTERED INDEX [index_ActivityType] ON [{schema}].".BuildSchema(Schema) + $"[{tableName}]([Id] DESC);",
               _ => string.Empty
           };
    }
}
CompanyRepository

其次创建事务管理UnitOfWork

using System.Data;

namespace API.Dapper.UnitOfWork.V2
{
    public class UnitOfWorkV2 : IUnitOfWork
    {
        public IDbConnection DbConnection { get; set; }
        public IDbTransaction DbTransaction { get; set; }

        public UnitOfWorkV2(IDbConnectionSchema dbConnectionSchema)
        {
            DbConnection = dbConnectionSchema.DbConnection;
            DbConnection.Open();
            DbTransaction = dbConnectionSchema.DbConnection.BeginTransaction();
        }

        public UnitOfWorkV2(IDbConnectionSchema dbConnectionSchema, IsolationLevel isolationLevel)
        {
            DbConnection = dbConnectionSchema.DbConnection;
            DbConnection.Open();
            DbTransaction = dbConnectionSchema.DbConnection.BeginTransaction(isolationLevel);
        }

        public void Commit()
        {
            DbTransaction.Commit();
        }

        public void Dispose()
        {
            if (DbTransaction != null)
                DbTransaction.Dispose();
            if (DbConnection != null) 
                DbConnection.Dispose();
            GC.SuppressFinalize(this);
        }

        public void Rollback()
        {
            DbTransaction.Rollback();
        }
    }
}
UnitOfWork 
using API.Dapper.Repository.V2;

namespace API.Dapper.UnitOfWork.V2;

public class CompanyUnitOfWorkV2 : UnitOfWorkV2
{
    public CompanyRepositoryV2 CompanyRepositoryV2 { get; }
    public CompanyUnitOfWorkV2(IDbConnectionSchema dbConnectionSchema) 
        : base(dbConnectionSchema)
    {
        CompanyRepositoryV2 = new CompanyRepositoryV2(dbConnectionSchema, DbTransaction);
    }

}
CompanyUnitOfWork

IDbConnection扩展方法

 #region Partition
    public static async Task<bool> IsPartitionTable(this IDbConnection connection, string tableName, string schema, IDbTransaction transaction = null, int? commandTimeout = null)
    {
        var sql = $"SELECT Count(t.object_id) FROM sys.tables t INNER JOIN sys.indexes i ON t.object_id = i.object_id AND i.type IN(0,1) INNER JOIN sys.partition_schemes ps ON i.data_space_id = ps.data_space_id WHERE t.name = '{tableName}' AND SCHEMA_NAME(schema_id) = '{schema}';";
        return await connection.ExecuteScalarAsync<int>(sql, transaction: transaction, commandTimeout: commandTimeout) > 0;
    }

    public static async Task<Int32> CreatePartitionValues(this IDbConnection connection, IEnumerable<string> rangeValues, string tableName, string schema, IDbTransaction transaction = null, int? commandTimeout = null)
    {
        var partitionSchemaFunctions = await connection.GetPartitionSchemaFunctions(tableName, schema, transaction, commandTimeout);
        if (partitionSchemaFunctions?.Count() == 0)
            throw new Exception($"The partition scheme and function is null or empty, target table: {tableName}, schema: {schema}.");
        var partition = partitionSchemaFunctions!.First();
        var affectedRows = 0;
        foreach (var rangeValue in rangeValues)
        {
            var sql = $"ALTER PARTITION SCHEME [{partition.Scheme}] Next used 'PRIMARY' ALTER PARTITION FUNCTION [{partition.Function}]() SPLIT RANGE('{rangeValue}');";
            affectedRows += await connection.ExecuteAsync(sql, transaction: transaction, commandTimeout: commandTimeout);
        }
        return affectedRows;
    }

    private static async Task<IEnumerable<PartitionEntity>> GetPartitionSchemaFunctions(this IDbConnection connection, string tableName, string schema, IDbTransaction transaction = null, int? commandTimeout = null)
    {
        var sql = $"SELECT ps.name [Scheme], pf.name [Function] FROM sys.partition_functions AS pf JOIN sys.partition_schemes AS ps ON pf.function_id = ps.function_id JOIN sys.indexes AS si ON ps.data_space_id = si.data_space_id AND si.type IN(0,1) JOIN sys.tables AS st ON si.object_id = st.object_id WHERE st.name = '{tableName}'  AND SCHEMA_NAME(schema_id) = '{schema}'";
        return await connection.QueryAsync<PartitionEntity>(sql, transaction: transaction, commandTimeout: commandTimeout);
    }

    public static async Task<IEnumerable<string>> GetPartitionValues(this IDbConnection connection, string tableName, string schema, IDbTransaction transaction = null, int? commandTimeout = null)
    {
        var sql = $"SELECT srv.value FROM sys.tables AS st JOIN sys.indexes AS si ON st.object_id = si.object_id JOIN sys.partitions AS sp ON si.object_id = sp.object_id AND si.index_id = sp.index_id JOIN sys.partition_schemes AS ps ON si.data_space_id = ps.data_space_id JOIN sys.partition_functions AS sf ON ps.function_id = sf.function_id JOIN sys.partition_range_values AS srv ON sf.function_id = srv.function_id and srv.boundary_id = sp.partition_number WHERE si.type <= 1 AND st.name = '{tableName}' AND SCHEMA_NAME(schema_id) = '{schema}'";
        return await connection.QueryAsync<string>(sql, transaction: transaction, commandTimeout: commandTimeout);
    }
    #endregion
DbConnectionExtension

sql

namespace API.Dapper;

public static partial class Sql
{
    public const string Insert = @"INSERT INTO [{schema}].[{0}]({1})VALUES({2})";
    public const string Select = @"SELECT {0} FROM [{schema}].{1} {2}";
    public const string Update = @"UPDATE [{schema}].[{0}] SET {1} {2}";
    public const string Delete = @"DELETE FROM [{schema}].[{0}] {1}";

    public static string BuildSchema(this string sql, string schema)
        => sql.Replace("{schema}", schema);
}
Sql

创建一个NetCore Console Application，引入上面Library工程，定义ComanyService

using API.Dapper.Entity;
using API.Dapper.UnitOfWork.V2;

namespace API.Dapper.Test.Services;

public class CompanyService
{
    public async Task<IEnumerable<Company>> GetAllAsync()
    {
        using var unitOfWork = new CompanyUnitOfWorkV2(new DemoDbConnectionSchema());
        return await unitOfWork.CompanyRepositoryV2.GetAllAsync();
    }

    public async Task<Company> GetAsync(Guid id)
    {
        using var unitOfWork = new CompanyUnitOfWorkV2(new DemoDbConnectionSchema());
        return await unitOfWork.CompanyRepositoryV2.GetAsync(id);
    }

    public async Task AddAsync(IEnumerable<Company> companies)
    {
        using var unitOfWork = new CompanyUnitOfWorkV2(new DemoDbConnectionSchema());
        await unitOfWork.CompanyRepositoryV2.AddAsync(companies);
        unitOfWork.Commit();
    }
}
CompanyService

在Program中通过DI调用UserService测试AddAsync及GetAsync

//DI
var services = new ServiceCollection();
services.AddTransient<CompanyService>();
var serviceProvider = services.BuildServiceProvider();

UserContext.Current.Init(new UserPrincipal
{
    UserId = Guid.Parse("00000000-0000-0000-0000-000000000001"),
    DisplayName = "Jack",
    Role = Role.Boss,
    HireDate = DateTime.UtcNow.Ticks
});

var companyService = serviceProvider.GetRequiredService<CompanyService>();
Console.WriteLine("-----------Start to add company to partition table--------------");
await companyService.AddAsync(new List<Company>
{
    new Company { Id = Guid.NewGuid(), Name ="Compnay01", Region ="AU" },
    new Company { Id= Guid.NewGuid(), Name ="Company02",Region="US"}
});
Console.WriteLine("-----------Add two companies to table----------------------------");
var companies = await companyService.GetAllAsync();
Console.WriteLine($"Get all users from tables \r\n {string.Join("\r\n", companies.Select(c => $"Id: {c.Id}, Name: {c.Name}, Region: {c.Region}"))}");
Console.ReadKey();
Program

运行结果

 查看DB table及分区函数

 分区Scheme及Function

 在database storage中查看数据分布情况

 OK，简单的分区即完成，但在真正的生产环境通常不会指定同一个data file[Primary]作为分区,可以指定多个data file，也可能会将多个data file放在不同磁盘上，从而提高IO。当前资源和精力有限，之后会有需要会逐渐完善。

标签：string,NetCore,分区,tableName,id,sql,var,Dapper,public
From： https://www.cnblogs.com/qindy/p/17838857.html

相关文章

  • OTA A/B 分区升级 update_engine简介update_engine简介
    近期猛然发现公司的项目都已经换成了AB升级，AB升级之前一直有所了解，只是一直都没有去仔细查看过其具体升级流程，这两天抽空捋了捋，简单整理下。AB升级（谷歌官网叫法无缝更新）是自android7.0开始新增的一种android设备升级方式，只是这种方式对设备存储要求高，所以之前没有怎么被国内手机厂......
  • Dapper QueryMultiple throws "No columns were selected"
    调试存储过程发现某字段NULL，拼成的SQL最后啥也没有了。。。原因是存储过程没有执行到查询SQL语句QueryAsyncthrows"Nocolumnswereselected"whenstoredprocdoesn'tperformaselectstatement #591Closed  toannguyen83 openedthisissue onAug27,20......
  • CentOS7安装强制使用gpt分区
    安装CentOS时，系统默认会使用mbr模式。如果磁盘大于2T，则需要如下操作，在开机启动画面按tab键，以便输入核心参数，在行末输入空格后inst.gpt，回车： 可以如下模式分区，其中根分区既可以是基本的xfs或者ext3，也可以是lvm格式： ......
  • .NetCore Ocelot
    参考：https://blog.51cto.com/u_13746169/5876532Nuget包：OcelotOcelot.Provider.ConsulOcelot.Provider.PollyOcelot.Cache.CacheManager 程序包管理器控制台CLIInstall-PackageOcelot-Version16.0.1Install-PackageOcelot.Provider.Consul-Version16.0.1Install-Packag......
  • 持续集成指南：GitHubAction 自动构建+部署AspNetCore项目
    前言之前研究了使用GitHubAction自动构建和发布nuget包：开发现代化的.NetCore控制台程序：(4)使用GithubAction自动构建以及发布nuget包现在更进一步，使用GitHubAction在其提供的runner里构建docker镜像，之后提交到阿里云镜像私有仓库，再在本地的runner将镜像pull下来......
  • netcore net 递归查询示例
    ///<summary>///查询项目列表///</summary>///<paramname="userModel"></param>///<returns></returns>publicasyncTask<List<GetProjectListOutput>>GetProjectList......
  • mysql-partition库-创建数据库表分区、展示表分区最新字段
    说明[tips]提供mysql数据库的分区表最新分区查询，支持分区字段为DATE或TO_DAYS(date_str)类型的数据表新建分区,依赖库包括："pandas","pymysql","sqlalchemy".[ProvidethelatestpartitionqueryforthepartitiontableofMySQLdatabase,supportingpartitionfields......
  • centos7 关闭swap分区
    目录1.查看swap分区信息2.关闭swap分区 1.查看swap分区信息通过free-h命令查看[root@k8s-master-02~]#free-htotalusedfreesharedbuff/cacheavailableMem:1.8G131M1.4G8.9M212M1.5GSwap:2.0G0B2.0G 2.关闭swap分区2.1临时关闭（ swapoff-a ）[r......
  • C++ 内存分区
    C/C++内存管理C/C++内存分布转载https://www.coonote.com/note/cpp-memory-management.htmlC/C++程序内存分配的几个区域：栈区（stack）在执行函数时，函数内局部变量的存储单元都可以在栈上创建，函数执行结束时这些存储单元自动被释放。栈内存分配运算内置于处理器的指令集......
  • Linux分区
    转自Linux硬盘分区指南-知乎(zhihu.com)本文不涉及具体工具的操作和使用流程，仅从概念方面阐述Linux分区到挂载所牵涉的相关知识点，适合Linux新人快速建立对相关知识的概念和认知。分区和目录和Windows使用盘符定义分区的方式不同，Linux系统使用树状目录的方式来定义分......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99