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

支持abp.vnext

This commit is contained in:
xuejiaming 2021-10-22 17:26:02 +08:00
parent 6539e1e74a
commit 3963d3bd0f
6 changed files with 499 additions and 365 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

825
samples/Samples.AbpSharding/AbstractShardingAbpDbContext.cs
View File

@ -1,400 +1,519 @@
//using System;
//using System.Collections.Concurrent;
//using System.Collections.Generic;
//using System.Data;
//using System.Data.Common;
//using System.Linq;
//using System.Linq.Expressions;
//using System.Threading;
//using System.Threading.Tasks;
//using Microsoft.EntityFrameworkCore;
//using Microsoft.EntityFrameworkCore.ChangeTracking;
//using Microsoft.EntityFrameworkCore.Storage;
//using ShardingCore;
//using ShardingCore.Core;
//using ShardingCore.Core.VirtualRoutes.TableRoutes.RouteTails.Abstractions;
//using ShardingCore.Extensions;
//using ShardingCore.Sharding.Abstractions;
//using ShardingCore.Sharding.ShardingDbContextExecutors;
//using ShardingCore.Sharding.ShardingTransactions;
//using Volo.Abp.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.ChangeTracking;
using ShardingCore.Core.VirtualRoutes.TableRoutes.RouteTails.Abstractions;
using ShardingCore.Extensions;
using ShardingCore.Sharding.Abstractions;
using ShardingCore.Sharding.ShardingDbContextExecutors;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using ShardingCore.EFCores.OptionsExtensions;
using Volo.Abp.EntityFrameworkCore;
//namespace Samples.AbpSharding
//{
// public abstract class AbstractShardingAbpDbContext : AbpDbContext<AbstractShardingAbpDbContext>, IShardingDbContext, ISupportShardingTransaction, ISupportShardingReadWrite
// {
// private readonly IShardingDbContextExecutor _shardingDbContextExecutor;
// protected AbstractShardingAbpDbContext(DbContextOptions<AbstractShardingAbpDbContext> options) : base(options)
// {
namespace Samples.AbpSharding
{
public abstract class AbstractShardingAbpDbContext : AbpDbContext<AbstractShardingAbpDbContext>, IShardingDbContext, ISupportShardingTransaction, ISupportShardingReadWrite
{
private readonly IShardingDbContextExecutor _shardingDbContextExecutor;
protected AbstractShardingAbpDbContext(DbContextOptions<AbstractShardingAbpDbContext> options) : base(options)
{
// _shardingDbContextExecutor =
// (IShardingDbContextExecutor)Activator.CreateInstance(
// typeof(ShardingDbContextExecutor<>).GetGenericType0(this.GetType()));
// }
var wrapOptionsExtension = options.FindExtension<ShardingWrapOptionsExtension>();
if (wrapOptionsExtension != null)
{
_shardingDbContextExecutor =
(IShardingDbContextExecutor)Activator.CreateInstance(
typeof(ShardingDbContextExecutor<>).GetGenericType0(this.GetType()), this);
}
}
// /// <summary>
// /// 读写分离优先级
// /// </summary>
// public int ReadWriteSeparationPriority
// {
// get => _shardingDbContextExecutor.ReadWriteSeparationPriority;
// set => _shardingDbContextExecutor.ReadWriteSeparationPriority = value;
// }
// /// <summary>
// /// 是否使用读写分离
// /// </summary>
// public bool ReadWriteSeparation
// {
// get => _shardingDbContextExecutor.ReadWriteSeparation;
// set => _shardingDbContextExecutor.ReadWriteSeparation = value;
// }
/// <summary>
/// 读写分离优先级
/// </summary>
public int ReadWriteSeparationPriority
{
get => _shardingDbContextExecutor.ReadWriteSeparationPriority;
set => _shardingDbContextExecutor.ReadWriteSeparationPriority = value;
}
/// <summary>
/// 是否使用读写分离
/// </summary>
public bool ReadWriteSeparation
{
get => _shardingDbContextExecutor.ReadWriteSeparation;
set => _shardingDbContextExecutor.ReadWriteSeparation = value;
}
// public new bool IsExecutor { get; private set; }
/// <summary>
/// 是否是真正的执行者
/// </summary>
private bool isExecutor => _shardingDbContextExecutor == null;
//public void ShardingUpgrade()
//{
// IsExecutor = true;
// //isExecutor = true;
//}
// public DbContext GetDbContext(string dataSourceName, bool parallelQuery, IRouteTail routeTail)
// {
// var dbContext = _shardingDbContextExecutor.CreateDbContext(parallelQuery, dataSourceName, routeTail);
// if (!parallelQuery)
// ((AbpDbContext<AbstractShardingAbpDbContext>)dbContext).LazyServiceProvider = this.LazyServiceProvider;
// return dbContext;
// }
public DbContext GetDbContext(string dataSourceName, bool parallelQuery, IRouteTail routeTail)
{
var dbContext = _shardingDbContextExecutor.CreateDbContext(parallelQuery, dataSourceName, routeTail);
if (!parallelQuery&& dbContext is AbstractShardingAbpDbContext abstractShardingAbpDbContext)
{
abstractShardingAbpDbContext.LazyServiceProvider = this.LazyServiceProvider;
}
// /// <summary>
// /// 根据对象创建通用的dbcontext
// /// </summary>
// /// <typeparam name="TEntity"></typeparam>
// /// <param name="entity"></param>
// /// <returns></returns>
// public DbContext CreateGenericDbContext<TEntity>(TEntity entity) where TEntity : class
// {
// return _shardingDbContextExecutor.CreateGenericDbContext(entity);
// }
return dbContext;
}
/// <summary>
/// 根据对象创建通用的dbcontext
/// </summary>
/// <typeparam name="TEntity"></typeparam>
/// <param name="entity"></param>
/// <returns></returns>
public DbContext CreateGenericDbContext<TEntity>(TEntity entity) where TEntity : class
{
return _shardingDbContextExecutor.CreateGenericDbContext(entity);
}
// public override EntityEntry Add(object entity)
// {
// return CreateGenericDbContext(entity).Add(entity);
// }
public override EntityEntry Add(object entity)
{
if (isExecutor)
base.Add(entity);
return CreateGenericDbContext(entity).Add(entity);
}
// public override EntityEntry<TEntity> Add<TEntity>(TEntity entity)
// {
// return CreateGenericDbContext(entity).Add(entity);
// }
public override EntityEntry<TEntity> Add<TEntity>(TEntity entity)
{
if (isExecutor)
return base.Add(entity);
return CreateGenericDbContext(entity).Add(entity);
}
public override ValueTask<EntityEntry<TEntity>> AddAsync<TEntity>(TEntity entity, CancellationToken cancellationToken = new CancellationToken())
{
if (isExecutor)
return base.AddAsync(entity, cancellationToken);
return CreateGenericDbContext(entity).AddAsync(entity, cancellationToken);
}
public override ValueTask<EntityEntry> AddAsync(object entity, CancellationToken cancellationToken = new CancellationToken())
{
if (isExecutor)
return base.AddAsync(entity, cancellationToken);
return CreateGenericDbContext(entity).AddAsync(entity, cancellationToken);
}
public override void AddRange(params object[] entities)
{
if (isExecutor)
{
base.AddRange(entities);
return;
}
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)
{
if (isExecutor)
{
base.AddRange(entities);
return;
}
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)
{
if (isExecutor)
{
await base.AddRangeAsync(entities);
return;
}
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())
{
if (isExecutor)
{
await base.AddRangeAsync(entities, cancellationToken);
return;
}
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)
{
if (isExecutor)
return base.Attach(entity);
return CreateGenericDbContext(entity).Attach(entity);
}
public override EntityEntry Attach(object entity)
{
if (isExecutor)
return base.Attach(entity);
return CreateGenericDbContext(entity).Attach(entity);
}
public override void AttachRange(params object[] entities)
{
if (isExecutor)
{
base.AttachRange(entities);
return;
}
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)
{
if (isExecutor)
{
base.AttachRange(entities);
return;
}
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 ValueTask<EntityEntry<TEntity>> AddAsync<TEntity>(TEntity entity, CancellationToken cancellationToken = new CancellationToken())
// {
// return CreateGenericDbContext(entity).AddAsync(entity, cancellationToken);
// }
//public override DatabaseFacade Database => _dbContextCaches.Any()
// ? _dbContextCaches.First().Value.Database
// : GetDbContext(true, string.Empty).Database;
// public override ValueTask<EntityEntry> AddAsync(object entity, CancellationToken cancellationToken = new CancellationToken())
// {
// return CreateGenericDbContext(entity).AddAsync(entity, cancellationToken);
// }
public override EntityEntry<TEntity> Entry<TEntity>(TEntity entity)
{
if (isExecutor)
return base.Entry(entity);
return CreateGenericDbContext(entity).Entry(entity);
}
// 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);
public override EntityEntry Entry(object entity)
{
if (isExecutor)
return base.Entry(entity);
return CreateGenericDbContext(entity).Entry(entity);
}
// foreach (var group in groups)
// {
// group.Key.AddRange(group.Select(o => o.Entity));
// }
// }
public override EntityEntry<TEntity> Update<TEntity>(TEntity entity)
{
if (isExecutor)
return base.Update(entity);
return CreateGenericDbContext(entity).Update(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);
public override EntityEntry Update(object entity)
{
if (isExecutor)
return base.Update(entity);
return CreateGenericDbContext(entity).Update(entity);
}
// foreach (var group in groups)
// {
// group.Key.AddRange(group.Select(o => o.Entity));
// }
// }
public override void UpdateRange(params object[] entities)
{
if (isExecutor)
{
base.UpdateRange(entities);
return;
}
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
// 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)
{
group.Key.UpdateRange(group.Select(o => o.Entity));
}
}
// foreach (var group in groups)
// {
// await group.Key.AddRangeAsync(group.Select(o => o.Entity));
// }
// }
public override void UpdateRange(IEnumerable<object> entities)
{
if (isExecutor)
{
base.UpdateRange(entities);
return;
}
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
// 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)
{
group.Key.UpdateRange(group.Select(o => o.Entity));
}
}
// foreach (var group in groups)
// {
// await group.Key.AddRangeAsync(group.Select(o => o.Entity));
// }
// }
public override EntityEntry<TEntity> Remove<TEntity>(TEntity entity)
{
if (isExecutor)
return base.Remove(entity);
return CreateGenericDbContext(entity).Remove(entity);
}
// public override EntityEntry<TEntity> Attach<TEntity>(TEntity entity)
// {
// return CreateGenericDbContext(entity).Attach(entity);
// }
public override EntityEntry Remove(object entity)
{
if (isExecutor)
return base.Remove(entity);
return CreateGenericDbContext(entity).Remove(entity);
}
// public override EntityEntry Attach(object entity)
// {
// return CreateGenericDbContext(entity).Attach(entity);
// }
public override void RemoveRange(params object[] entities)
{
if (isExecutor)
{
base.RemoveRange(entities);
return;
}
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
// 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.RemoveRange(group.Select(o => o.Entity));
}
}
// foreach (var group in groups)
// {
// group.Key.AttachRange(group.Select(o => o.Entity));
// }
// }
public override void RemoveRange(IEnumerable<object> entities)
{
if (isExecutor)
{
base.RemoveRange(entities);
return;
}
var groups = entities.Select(o =>
{
var dbContext = CreateGenericDbContext(o);
return new
{
DbContext = dbContext,
Entity = o
};
}).GroupBy(g => g.DbContext);
// 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.RemoveRange(group.Select(o => o.Entity));
}
}
// foreach (var group in groups)
// {
// group.Key.AttachRange(group.Select(o => o.Entity));
// }
// }
public override int SaveChanges()
{
if (isExecutor)
return base.SaveChanges();
return this.SaveChanges(true);
}
public override int SaveChanges(bool acceptAllChangesOnSuccess)
{
if (isExecutor)
return base.SaveChanges(acceptAllChangesOnSuccess);
//ApplyShardingConcepts();
int i = 0;
//如果是内部开的事务就内部自己消化
if (Database.CurrentTransaction == null && _shardingDbContextExecutor.IsMultiDbContext)
{
using (var tran = Database.BeginTransaction())
{
i = _shardingDbContextExecutor.SaveChanges(acceptAllChangesOnSuccess);
tran.Commit();
}
}
else
{
i = _shardingDbContextExecutor.SaveChanges(acceptAllChangesOnSuccess);
}
return i;
}
// //public override DatabaseFacade Database => _dbContextCaches.Any()
// // ? _dbContextCaches.First().Value.Database
// // : GetDbContext(true, string.Empty).Database;
public override Task<int> SaveChangesAsync(CancellationToken cancellationToken = new CancellationToken())
{
if (isExecutor)
return base.SaveChangesAsync(cancellationToken);
return this.SaveChangesAsync(true, cancellationToken);
}
// public override EntityEntry<TEntity> Entry<TEntity>(TEntity entity)
// {
// return CreateGenericDbContext(entity).Entry(entity);
// }
public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = new CancellationToken())
{
if (isExecutor)
return await base.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
//ApplyShardingConcepts();
int i = 0;
//如果是内部开的事务就内部自己消化
if (Database.CurrentTransaction == null && _shardingDbContextExecutor.IsMultiDbContext)
{
using (var tran = await Database.BeginTransactionAsync(cancellationToken))
{
i = await _shardingDbContextExecutor.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
// 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()
// {
// return this.SaveChanges(true);
// }
// public override int SaveChanges(bool acceptAllChangesOnSuccess)
// {
// //ApplyShardingConcepts();
// int i = 0;
// //如果是内部开的事务就内部自己消化
// if (!_shardingDbContextExecutor.IsBeginTransaction)
// {
// using (var tran = _shardingDbContextExecutor.BeginTransaction())
// {
// i = _shardingDbContextExecutor.SaveChanges(acceptAllChangesOnSuccess);
// tran.Commit();
// }
// }
// else
// {
// i = _shardingDbContextExecutor.SaveChanges(acceptAllChangesOnSuccess);
// }
// return i;
// }
await tran.CommitAsync(cancellationToken);
}
}
else
{
i = await _shardingDbContextExecutor.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
}
// public override Task<int> SaveChangesAsync(CancellationToken cancellationToken = new CancellationToken())
// {
// return this.SaveChangesAsync(true, cancellationToken);
// }
return i;
}
// public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = new CancellationToken())
// {
// //ApplyShardingConcepts();
// int i = 0;
// //如果是内部开的事务就内部自己消化
// if (!_shardingDbContextExecutor.IsBeginTransaction)
// {
// using (var tran = _shardingDbContextExecutor.BeginTransaction())
// {
// i = await _shardingDbContextExecutor.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
public override void Dispose()
{
// await tran.CommitAsync(cancellationToken);
// }
// }
// else
// {
// i = await _shardingDbContextExecutor.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
// }
if (isExecutor)
{
base.Dispose();
}
else
{
_shardingDbContextExecutor.Dispose();
base.Dispose();
}
}
public override async ValueTask DisposeAsync()
{
if (isExecutor)
{
await base.DisposeAsync();
}
else
{
await _shardingDbContextExecutor.DisposeAsync();
// return i;
// }
await base.DisposeAsync();
}
}
public Task RollbackAsync(CancellationToken cancellationToken = new CancellationToken())
{
return _shardingDbContextExecutor.RollbackAsync(cancellationToken);
}
// public override void Dispose()
// {
// _shardingDbContextExecutor.Dispose();
// base.Dispose();
// }
public Task CommitAsync(CancellationToken cancellationToken = new CancellationToken())
{
return _shardingDbContextExecutor.CommitAsync(cancellationToken);
}
// public override async ValueTask DisposeAsync()
// {
// await _shardingDbContextExecutor.DisposeAsync();
public void NotifyShardingTransaction()
{
_shardingDbContextExecutor.NotifyShardingTransaction();
}
// await base.DisposeAsync();
// }
public void Rollback()
{
_shardingDbContextExecutor.Rollback();
}
// public IShardingTransaction BeginTransaction(IsolationLevel isolationLevel = IsolationLevel.Unspecified)
// {
// return _shardingDbContextExecutor.BeginTransaction(isolationLevel);
// }
// }
//}
public void Commit()
{
_shardingDbContextExecutor.Commit();
}
}
}

6
src/ShardingCore/Sharding/AbstractShardingDbContext.cs
View File

@ -468,7 +468,7 @@ namespace ShardingCore.Sharding
//ApplyShardingConcepts();
int i = 0;
//如果是内部开的事务就内部自己消化
if (Database.CurrentTransaction==null)
if (Database.CurrentTransaction==null&&_shardingDbContextExecutor.IsMultiDbContext)
{
using (var tran = Database.BeginTransaction())
{
@ -499,9 +499,9 @@ namespace ShardingCore.Sharding
//ApplyShardingConcepts();
int i = 0;
//如果是内部开的事务就内部自己消化
if (Database.CurrentTransaction==null)
if (Database.CurrentTransaction==null && _shardingDbContextExecutor.IsMultiDbContext)
{
using (var tran = Database.BeginTransaction())
using (var tran = await Database.BeginTransactionAsync(cancellationToken))
{
i = await _shardingDbContextExecutor.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
#if EFCORE2

4
src/ShardingCore/Sharding/Abstractions/IShardingDbContextExecutor.cs
View File

@ -28,6 +28,10 @@ namespace ShardingCore.Sharding.Abstractions
/// 当前是否开启读写分离
/// </summary>
bool ReadWriteSeparation { get; set; }
/// <summary>
/// 是否存在多个dbcontext
/// </summary>
bool IsMultiDbContext { get; }
/// <summary>

9
src/ShardingCore/Sharding/ShardingDbContextExecutors/DataSourceDbContext.cs
View File

@ -28,6 +28,7 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
public class DataSourceDbContext<TShardingDbContext> : IDataSourceDbContext where TShardingDbContext : DbContext, IShardingDbContext
{
public bool IsDefault { get; }
public int DbContextCount => _dataSourceDbContexts.Count;
private readonly IShardingDbContextOptionsBuilderConfig<TShardingDbContext> _shardingDbContextOptionsBuilderConfig;
private readonly IShardingDbContextFactory<TShardingDbContext> _shardingDbContextFactory;
private readonly ActualConnectionStringManager<TShardingDbContext> _actualConnectionStringManager;
@ -252,7 +253,7 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
}
}
public void Commit()
public void Commit(int dataSourceCount)
{
if (IsDefault)
return;
@ -263,6 +264,8 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
catch (Exception e)
{
_logger.LogError(e, "commit error.");
if (dataSourceCount == 1)
throw;
}
}
#if !EFCORE2
@ -283,7 +286,7 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
}
}
public async Task CommitAsync(CancellationToken cancellationToken = new CancellationToken())
public async Task CommitAsync(int dataSourceCount,CancellationToken cancellationToken = new CancellationToken())
{
cancellationToken.ThrowIfCancellationRequested();
if (IsDefault)
@ -296,6 +299,8 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
catch (Exception e)
{
_logger.LogError(e, "commit error.");
if (dataSourceCount == 1)
throw;
}
}
#endif

5
src/ShardingCore/Sharding/ShardingDbContextExecutors/IDataSourceDbContext.cs
View File

@ -24,6 +24,7 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
#endif
{
bool IsDefault { get; }
int DbContextCount { get; }
DbContext CreateDbContext(IRouteTail routeTail);
void NotifyTransaction();
@ -34,10 +35,10 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
void Rollback();
void Commit();
void Commit(int dataSourceCount);
#if !EFCORE2
Task RollbackAsync(CancellationToken cancellationToken = new CancellationToken());
Task CommitAsync(CancellationToken cancellationToken = new CancellationToken());
Task CommitAsync(int dataSourceCount,CancellationToken cancellationToken = new CancellationToken());
#endif
}

9
src/ShardingCore/Sharding/ShardingDbContextExecutors/ShardingDbContextExecutor.cs
View File

@ -76,6 +76,11 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
return _dbContextCaches.GetOrAdd(dataSourceName, dsname => new DataSourceDbContext<TShardingDbContext>(dataSourceName, _virtualDataSource.IsDefault(dataSourceName), _shardingDbContext, _shardingDbContextOptionsBuilderConfig, _shardingDbContextFactory, _actualConnectionStringManager));
}
/// <summary>
/// 是否有多个dcontext
/// </summary>
public bool IsMultiDbContext =>
_dbContextCaches.Count > 1 || _dbContextCaches.Sum(o => o.Value.DbContextCount) > 1;
public DbContext CreateDbContext(bool parallelQuery, string dataSourceName, IRouteTail routeTail)
{
@ -155,7 +160,7 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
{
foreach (var dbContextCache in _dbContextCaches)
{
dbContextCache.Value.Commit();
dbContextCache.Value.Commit(_dbContextCaches.Count);
}
}
@ -192,7 +197,7 @@ namespace ShardingCore.Sharding.ShardingDbContextExecutors
{
foreach (var dbContextCache in _dbContextCaches)
{
await dbContextCache.Value.CommitAsync(cancellationToken);
await dbContextCache.Value.CommitAsync(_dbContextCaches.Count,cancellationToken);
}
}
public async ValueTask DisposeAsync()