Autotest
/
sharding
mirror of https://gitee.com/dotnetchina/sharding-core.git
1
0
Fork 0
Code Issues Releases Wiki Activity

添加内部版事务功能

This commit is contained in:
xuejiaming 2021-08-15 21:31:59 +08:00
parent bb8a6e3a0f
commit 9a53dcd684
11 changed files with 532 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/ShardingCore/DbContexts/ShardingDbContextFactory.cs
View File

@ -34,22 +34,13 @@ namespace ShardingCore.DbContexts
}
public DbContext Create(ShardingDbContextOptions shardingDbContextOptions)
{
var tail=shardingDbContextOptions.Tail;
var shardingConfigEntry = _shardingCoreOptions.GetShardingConfig();
using (var scope = _shardingTableScopeFactory.CreateScope())
var dbContext = shardingConfigEntry.Creator(shardingDbContextOptions);
if (!string.IsNullOrWhiteSpace(tail) && dbContext is IShardingTableDbContext shardingTableDbContext)
{
string tail = null;
string modelChangeKey = null;
if (!string.IsNullOrWhiteSpace(shardingDbContextOptions.Tail))
{
tail = shardingDbContextOptions.Tail;
modelChangeKey = $"sharding_{tail}";
}
scope.ShardingTableAccessor.Context = ShardingTableContext.Create(tail);
var dbContext= shardingConfigEntry.Creator(shardingDbContextOptions);
if (modelChangeKey != null&& dbContext is IShardingTableDbContext shardingTableDbContext)
{
shardingTableDbContext.ModelChangeKey = modelChangeKey;
shardingTableDbContext.SetShardingTableDbContextTail(tail);
}
var filters = _dbContextCreateFilterManager.GetFilters();
@ -63,7 +54,6 @@ namespace ShardingCore.DbContexts
var dbContextModel = dbContext.Model;
return dbContext;
}
}
public DbContext Create(DbConnection dbConnection,string tail)
{

17
src/ShardingCore/DbContexts/ShardingDbContexts/IShardingTableAccessor.cs
View File

@ -1,17 +0,0 @@
namespace ShardingCore.DbContexts.ShardingDbContexts
{
/*
* @Author: xjm
* @Description:
* @Date: 2021/03/09 00:00:00
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
/// <summary>
///
/// </summary>
public interface IShardingTableAccessor
{
ShardingTableContext Context { get; set; }
}
}

21
src/ShardingCore/DbContexts/ShardingDbContexts/IShardingTableScopeFactory.cs
View File

@ -1,21 +0,0 @@
namespace ShardingCore.DbContexts.ShardingDbContexts
{
/*
* @Author: xjm
* @Description:
* @Date: 2021/03/09 00:00:00
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
/// <summary>
///
/// </summary>
public interface IShardingTableScopeFactory
{
/// <summary>
/// 创建查询scope
/// </summary>
/// <returns></returns>
ShardingTableScope CreateScope();
}
}

28
src/ShardingCore/DbContexts/ShardingDbContexts/ShardingTableAccessor.cs
View File

@ -1,28 +0,0 @@
using System.Threading;
using ShardingCore.Core.VirtualTables;
namespace ShardingCore.DbContexts.ShardingDbContexts
{
/*
* @Author: xjm
* @Description:
* @Date: 2021/03/09 13:08:15
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
/// <summary>
///
/// </summary>
public class ShardingTableAccessor:IShardingTableAccessor
{
private static AsyncLocal<ShardingTableContext> _context = new AsyncLocal<ShardingTableContext>();
/// <inheritdoc />
public ShardingTableContext Context
{
get => _context.Value;
set => _context.Value = value;
}
}
}

29
src/ShardingCore/DbContexts/ShardingDbContexts/ShardingTableContext.cs
View File

@ -1,29 +0,0 @@
namespace ShardingCore.DbContexts.ShardingDbContexts
{
/*
* @Author: xjm
* @Description:
* @Date: 2021/03/09 13:04:52
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
/// <summary>
/// 分表上下文
/// </summary>
public class ShardingTableContext
{
private ShardingTableContext(string tail)
{
Tail = tail;
}
public static ShardingTableContext Create(string tail)
{
return new ShardingTableContext(tail);
}
/// <summary>
/// 尾巴
/// </summary>
public string Tail { get; set; }
}
}

40
src/ShardingCore/DbContexts/ShardingDbContexts/ShardingTableScope.cs
View File

@ -1,40 +0,0 @@
using System;
namespace ShardingCore.DbContexts.ShardingDbContexts
{
/*
* @Author: xjm
* @Description:
* @Date: 2021/03/09 00:00:00
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
/// <summary>
///
/// </summary>
public class ShardingTableScope: IDisposable
{
/// <summary>
/// 分表配置访问器
/// </summary>
public IShardingTableAccessor ShardingTableAccessor { get; }
/// <summary>
/// 构造函数
/// </summary>
/// <param name="shardingAccessor"></param>
public ShardingTableScope(IShardingTableAccessor shardingAccessor)
{
ShardingTableAccessor = shardingAccessor;
}
/// <summary>
/// 回收
/// </summary>
public void Dispose()
{
ShardingTableAccessor.Context = null;
}
}
}

36
src/ShardingCore/DbContexts/ShardingDbContexts/ShardingTableScopeFactory.cs
View File

@ -1,36 +0,0 @@
namespace ShardingCore.DbContexts.ShardingDbContexts
{
/*
* @Author: xjm
* @Description:
* @Date: 2021/03/09 13:13:58
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
/// <summary>
///
/// </summary>
public class ShardingTableScopeFactory:IShardingTableScopeFactory
{
private readonly IShardingTableAccessor _shardingTableAccessor;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="shardingTableAccessor"></param>
public ShardingTableScopeFactory(IShardingTableAccessor shardingTableAccessor)
{
_shardingTableAccessor = shardingTableAccessor;
}
/// <summary>
/// 创建环境
/// </summary>
/// <returns></returns>
public ShardingTableScope CreateScope()
{
_shardingTableAccessor.Context = null;
return new ShardingTableScope(_shardingTableAccessor);
}
}
}

46
src/ShardingCore/Extensions/ShardingExtension.cs Normal file
View File

@ -0,0 +1,46 @@
using System;
using System.Collections.Generic;
using System.Text;
using ShardingCore.Exceptions;
using ShardingCore.Sharding.Abstractions;
namespace ShardingCore.Extensions
{
/*
* @Author: xjm
* @Description:
* @Date: 2021/8/15 16:12:27
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
/// <summary>
///
/// </summary>
public static class ShardingExtension
{
private static readonly string ShardingTableDbContextFormat = $"sharding_{Guid.NewGuid():n}_";
/// <summary>
/// 获取分表的tail
/// </summary>
/// <param name="dbContext"></param>
/// <returns></returns>
public static string GetShardingTableDbContextTail(this IShardingTableDbContext dbContext)
{
if (string.IsNullOrWhiteSpace(dbContext.ModelChangeKey))
throw new ShardingCoreException($"cant found ModelChangeKey in {dbContext.GetType().FullName}");
return dbContext.ModelChangeKey.Replace(ShardingTableDbContextFormat, string.Empty);
}
/// <summary>
/// 设置分表的tail
/// </summary>
/// <param name="dbContext"></param>
/// <param name="tail"></param>
public static void SetShardingTableDbContextTail(this IShardingTableDbContext dbContext, string tail)
{
if (!string.IsNullOrWhiteSpace(dbContext.ModelChangeKey))
throw new ShardingCoreException($"repeat set ModelChangeKey in {dbContext.GetType().FullName}");
dbContext.ModelChangeKey = $"{ShardingTableDbContextFormat}{tail}";
}
}
}

470
src/ShardingCore/Sharding/AbstractShardingDbContext.cs Normal file
View File

@ -0,0 +1,470 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.ChangeTracking;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Storage;
using ShardingCore.Core;
using ShardingCore.Core.VirtualRoutes.TableRoutes;
using ShardingCore.Core.VirtualTables;
using ShardingCore.DbContexts;
using ShardingCore.Extensions;
using ShardingCore.Sharding.Abstractions;
namespace ShardingCore.Sharding
{
/*
* @Author: xjm
* @Description:
* @Date: Saturday, 14 August 2021 09:57:08
* @Email: 326308290@qq.com
*/
/// <summary>
/// 分表分库的dbcontext
/// </summary>
/// <typeparam name="T"></typeparam>
public abstract class AbstractShardingDbContext<T> : DbContext, IShardingDbContext where T : DbContext
{
private readonly string EMPTY_SHARDING_TAIL_ID = Guid.NewGuid().ToString("n");
private readonly ConcurrentDictionary<string, DbContext> _dbContextCaches = new ConcurrentDictionary<string, DbContext>();
private readonly IVirtualTableManager _virtualTableManager;
private readonly IShardingDbContextFactory _shardingDbContextFactory;
public AbstractShardingDbContext(DbContextOptions options) : base(options)
{
_shardingDbContextFactory = ShardingContainer.GetService<IShardingDbContextFactory>();
}
public DbContext GetDbContext(bool track, string tail)
{
if (!_dbContextCaches.TryGetValue(tail, out var dbContext))
{
dbContext = _shardingDbContextFactory.Create(track ? this.Database.GetDbConnection() : null, tail == EMPTY_SHARDING_TAIL_ID ? string.Empty : tail); ;
_dbContextCaches.TryAdd(tail, dbContext);
}
//if (IsOpenTransaction)
//{
// _dbTransaction.Use(dbContext);
//}
return dbContext;
}
public bool IsBeginTransaction => Database.CurrentTransaction != null;
public DbContext CreateGenericDbContext<T>(T entity) where T : class
{
var tail = EMPTY_SHARDING_TAIL_ID;
if (entity.IsShardingTable())
{
var physicTable = _virtualTableManager.GetVirtualTable(entity.GetType()).RouteTo(new TableRouteConfig(null, entity as IShardingTable, null))[0];
tail = physicTable.Tail;
}
return GetDbContext(true, tail);
}
public override EntityEntry Add(object entity)
{
return CreateGenericDbContext(entity).Add(entity);
}
public override EntityEntry<TEntity> Add<TEntity>(TEntity entity)
{
return CreateGenericDbContext(entity).Add(entity);
}
public override ValueTask<EntityEntry<TEntity>> AddAsync<TEntity>(TEntity entity, CancellationToken cancellationToken = new CancellationToken())
{
return CreateGenericDbContext(entity).AddAsync(entity, cancellationToken);
}
public override ValueTask<EntityEntry> AddAsync(object entity, CancellationToken cancellationToken = new CancellationToken())
{
return CreateGenericDbContext(entity).AddAsync(entity, cancellationToken);
}
public override void AddRange(params object[] entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
group.Key.AddRange(group.Select(o => o.Entity));
}
}
public override void AddRange(IEnumerable<object> entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
group.Key.AddRange(group.Select(o => o.Entity));
}
}
public override async Task AddRangeAsync(params object[] entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
await group.Key.AddRangeAsync(group.Select(o => o.Entity));
}
}
public override async Task AddRangeAsync(IEnumerable<object> entities, CancellationToken cancellationToken = new CancellationToken())
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
await group.Key.AddRangeAsync(group.Select(o => o.Entity));
}
}
public override EntityEntry<TEntity> Attach<TEntity>(TEntity entity)
{
return CreateGenericDbContext(entity).Attach(entity);
}
public override EntityEntry Attach(object entity)
{
return CreateGenericDbContext(entity).Attach(entity);
}
public override void AttachRange(params object[] entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
group.Key.AttachRange(group.Select(o => o.Entity));
}
}
public override void AttachRange(IEnumerable<object> entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
group.Key.AttachRange(group.Select(o => o.Entity));
}
}
public override DatabaseFacade Database => _dbContextCaches.Any()
? _dbContextCaches.First().Value.Database
: GetDbContext(true, string.Empty).Database;
public override EntityEntry<TEntity> Entry<TEntity>(TEntity entity)
{
return CreateGenericDbContext(entity).Entry(entity);
}
public override EntityEntry Entry(object entity)
{
return CreateGenericDbContext(entity).Entry(entity);
}
public override EntityEntry<TEntity> Update<TEntity>(TEntity entity)
{
return CreateGenericDbContext(entity).Update(entity);
}
public override EntityEntry Update(object entity)
{
return CreateGenericDbContext(entity).Update(entity);
}
public override void UpdateRange(params object[] entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
group.Key.UpdateRange(group.Select(o => o.Entity));
}
}
public override void UpdateRange(IEnumerable<object> entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
group.Key.UpdateRange(group.Select(o => o.Entity));
}
}
public override EntityEntry<TEntity> Remove<TEntity>(TEntity entity)
{
return CreateGenericDbContext(entity).Remove(entity);
}
public override EntityEntry Remove(object entity)
{
return CreateGenericDbContext(entity).Remove(entity);
}
public override void RemoveRange(params object[] entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
group.Key.RemoveRange(group.Select(o => o.Entity));
}
}
public override void RemoveRange(IEnumerable<object> entities)
{
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
foreach (var group in groups)
{
group.Key.RemoveRange(group.Select(o => o.Entity));
}
}
public override int SaveChanges()
{
var isBeginTransaction = IsBeginTransaction;
//如果是内部开的事务就内部自己消化
if (!isBeginTransaction)
{
Database.BeginTransaction();
}
int i = 0;
try
{
foreach (var dbContextCache in _dbContextCaches)
{
dbContextCache.Value.Database.UseTransaction(Database.CurrentTransaction.GetDbTransaction());
i += dbContextCache.Value.SaveChanges();
}
if (!isBeginTransaction)
Database.CurrentTransaction.Commit();
}
finally
{
if (!isBeginTransaction)
Database.CurrentTransaction.Dispose();
}
return i;
}
public override int SaveChanges(bool acceptAllChangesOnSuccess)
{
var isBeginTransaction = IsBeginTransaction;
//如果是内部开的事务就内部自己消化
if (!isBeginTransaction)
{
Database.BeginTransaction();
}
int i = 0;
try
{
foreach (var dbContextCache in _dbContextCaches)
{
dbContextCache.Value.Database.UseTransaction(Database.CurrentTransaction.GetDbTransaction());
i += dbContextCache.Value.SaveChanges(acceptAllChangesOnSuccess);
}
if (!isBeginTransaction)
Database.CurrentTransaction.Commit();
}
finally
{
if (!isBeginTransaction)
Database.CurrentTransaction.Dispose();
}
return i;
}
public override async Task<int> SaveChangesAsync(CancellationToken cancellationToken = new CancellationToken())
{
var isBeginTransaction = IsBeginTransaction;
//如果是内部开的事务就内部自己消化
if (!isBeginTransaction)
{
await Database.BeginTransactionAsync(cancellationToken);
}
int i = 0;
try
{
foreach (var dbContextCache in _dbContextCaches)
{
await dbContextCache.Value.Database.UseTransactionAsync(Database.CurrentTransaction.GetDbTransaction(), cancellationToken: cancellationToken);
i +=await dbContextCache.Value.SaveChangesAsync(cancellationToken);
}
if (!isBeginTransaction)
await Database.CurrentTransaction.CommitAsync(cancellationToken);
}
finally
{
if (!isBeginTransaction)
await Database.CurrentTransaction.DisposeAsync();
}
return i;
}
public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = new CancellationToken())
{
var isBeginTransaction = IsBeginTransaction;
//如果是内部开的事务就内部自己消化
if (!isBeginTransaction)
{
await Database.BeginTransactionAsync(cancellationToken);
}
int i = 0;
try
{
foreach (var dbContextCache in _dbContextCaches)
{
await dbContextCache.Value.Database.UseTransactionAsync(Database.CurrentTransaction.GetDbTransaction(), cancellationToken: cancellationToken);
i += await dbContextCache.Value.SaveChangesAsync(acceptAllChangesOnSuccess,cancellationToken);
}
if (!isBeginTransaction)
await Database.CurrentTransaction.CommitAsync(cancellationToken);
}
finally
{
if (!isBeginTransaction)
await Database.CurrentTransaction.DisposeAsync();
}
return i;
}
public override void Dispose()
{
foreach (var dbContextCache in _dbContextCaches)
{
try
{
dbContextCache.Value.Dispose();
}
catch (Exception e)
{
Console.WriteLine(e);
}
}
base.Dispose();
}
public override async ValueTask DisposeAsync()
{
foreach (var dbContextCache in _dbContextCaches)
{
try
{
await dbContextCache.Value.DisposeAsync();
}
catch (Exception e)
{
Console.WriteLine(e);
}
}
await base.DisposeAsync();
}
}
}

61
src/ShardingCore/Sharding/ShardingDbContext.cs
View File

@ -1,61 +0,0 @@
using System;
using System.Collections.Concurrent;
using Microsoft.EntityFrameworkCore;
using ShardingCore.Core;
using ShardingCore.Core.VirtualRoutes.TableRoutes;
using ShardingCore.Core.VirtualTables;
using ShardingCore.DbContexts;
using ShardingCore.Extensions;
using ShardingCore.Sharding.Abstractions;
namespace ShardingCore.Sharding
{
/*
* @Author: xjm
* @Description:
* @Date: Saturday, 14 August 2021 09:57:08
* @Email: 326308290@qq.com
*/
/// <summary>
/// 分表分库的dbcontext
/// </summary>
/// <typeparam name="T"></typeparam>
public abstract class ShardingDbContext<T> : DbContext, IShardingDbContext where T : DbContext
{
private readonly string EMPTY_SHARDING_TAIL_ID = Guid.NewGuid().ToString("n");
private readonly ConcurrentDictionary<string, DbContext> _dbContextCaches = new ConcurrentDictionary<string, DbContext>();
private readonly IVirtualTableManager _virtualTableManager;
private readonly IShardingDbContextFactory _shardingDbContextFactory;
public ShardingDbContext(DbContextOptions options):base(options)
{
_shardingDbContextFactory = ShardingContainer.GetService<IShardingDbContextFactory>();
}
public DbContext GetDbContext(bool track, string tail)
{
if (!_dbContextCaches.TryGetValue(tail, out var dbContext))
{
dbContext = _shardingDbContextFactory.Create(track ? this.Database.GetDbConnection() : null, tail == EMPTY_SHARDING_TAIL_ID ? string.Empty : tail); ;
_dbContextCaches.TryAdd(tail, dbContext);
}
//if (IsOpenTransaction)
//{
// _dbTransaction.Use(dbContext);
//}
return dbContext;
}
public DbContext CreateGenericDbContext<T>(T entity) where T : class
{
var tail = EMPTY_SHARDING_TAIL_ID;
if (entity.IsShardingTable())
{
var physicTable = _virtualTableManager.GetVirtualTable(entity.GetType()).RouteTo(new TableRouteConfig(null, entity as IShardingTable, null))[0];
tail = physicTable.Tail;
}
return GetDbContext(true,tail);
}
}
}

2
test/ShardingCore.Test50/ShardingDefaultDbContext.cs
View File

@ -16,7 +16,7 @@ namespace ShardingCore.Test50
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
public class ShardingDefaultDbContext:ShardingDbContext<DefaultDbContext>
public class ShardingDefaultDbContext:AbstractShardingDbContext<DefaultDbContext>
{
public ShardingDefaultDbContext(DbContextOptions<ShardingDefaultDbContext> options) : base(options)
{