前言
简单整理一下工作单元模式。
正文
工作单元模式有3个特性，也算是其功能:
使用同一上下文
跟踪实体的状态
保障事务一致性
工作单元模式 主要关注事务，所以重点在事务上。
在共享层的基础建设类库中加入:
/// <summary>
/// 工作单元接口
/// </summary>
public interface IUnitOfWork : IDisposable
{
/// <summary>
/// 保存变更
/// </summary>
/// <param name="cancellationToken"></param>
/// <returns>返回受影响的数据条数</returns>
Task<int> SaveChangesAsync(CancellationToken cancellationToken = default);
/// <summary>
/// 保存变更
/// </summary>
/// <param name="cancellationToken"></param>
/// <returns>返回保存是否成功</returns>
Task<bool> SaveEntitiesAsync(CancellationToken cancellationToken = default);
}
SaveChangesAsync 事务第一个影响多少条数
SaveEntitiesAsync 事务是否成功
同样加入事务接口:
interface ITransaction
{
IDbContextTransaction GetCurrentTransaction();
bool HasActiveTransaction { get; }
Task<IDbContextTransaction> BeginTransactionAsync();
Task CommitTransactionAsync(IDbContextTransaction transaction);
void RollbackTransaction();
}
然后EFContext 实现它们:
/// <summary>
/// EF上下文
/// 注：在处理事务的逻辑部分，需要嵌入CAP的代码，构造函数参数 ICapPublisher
/// </summary>
public class EFContext : DbContext, IUnitOfWork, ITransaction
{
protected IMediator _mediator;
ICapPublisher _capBus;
public EFContext(DbContextOptions options, IMediator mediator, ICapPublisher capBus)
: base(options)
{
_mediator = mediator;
_capBus = capBus;
}
#region IUnitOfWork
/// <summary>
/// 保存实体变更
/// </summary>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async Task<bool> SaveEntitiesAsync(CancellationToken cancellationToken = default)
{
var result = await base.SaveChangesAsync(cancellationToken);
// 执行发送领域事件
await _mediator.DispatchDomainEventsAsync(this);
return true;
}
///// <summary>
///// IUniOfWork中该方法的定义与DbContext中的SaveChangesAsync一致，所以此处无需再进行实现
///// </summary>
///// <param name="cancellationToken"></param>
///// <returns></returns>
//public override Task<int> SaveChangesAsync(CancellationToken cancellationToken = default)
//{
// return base.SaveChangesAsync();
//}
#endregion
#region ITransaction
/// <summary>
/// 当前事务
/// </summary>
private IDbContextTransaction _currentTransaction;
/// <summary>
/// 公开方法，返回当前私有事务对象
/// </summary>
/// <returns></returns>
public IDbContextTransaction GetCurrentTransaction() => _currentTransaction;
/// <summary>
/// 当前事务是否开启
/// </summary>
public bool HasActiveTransaction => _currentTransaction == null;
/// <summary>
/// 开启事务
/// </summary>
/// <returns></returns>
public Task<IDbContextTransaction> BeginTransactionAsync()
{
if (_currentTransaction != null)
{
return null;
}
// 该扩展方法是由CAP组件提供
// 创建事务时，也要把 ICapPublisher 传入
// 核心作用是将我们要发送事件逻辑与我们业务的存储都放在同一个事务内部，从而保证事件与业务逻辑的存取都是一致的
_currentTransaction = Database.BeginTransaction(_capBus, autoCommit: false);
return Task.FromResult(_currentTransaction);
}
/// <summary>
/// 提交事务
/// </summary>
/// <param name="transaction"></param>
/// <returns></returns>
public async Task CommitTransactionAsync(IDbContextTransaction transaction)
{
if (transaction == null)
{
throw new ArgumentNullException(nameof(transaction));
}
if (transaction != _currentTransaction)
{
throw new InvalidOperationException($"Transaction {transaction.TransactionId} is not current");
}
try
{
// 提交事务之前，安全起见还是要 SaveChanges 一下，保存变更到数据库
await SaveChangesAsync();
transaction.Commit();
}
catch (Exception ex)
{
RollbackTransaction();
throw;
}
finally
{
if (_currentTransaction!=null)
{
_currentTransaction.Dispose();
_currentTransaction = null;
}
}
}
/// <summary>
/// 回滚事务
/// </summary>
public void RollbackTransaction()
{
try
{
_currentTransaction?.Rollback();
}
finally
{
if (_currentTransaction!=null)
{
_currentTransaction.Dispose();
_currentTransaction = null;
}
}
}
#endregion
}
前面这两个实现了工作单元模式的事务的功能，那么还有一个问题，如何实现管理我们的事务。
/// <summary>
/// 注入事务管理过程
/// </summary>
/// <typeparam name="TDbContext"></typeparam>
/// <typeparam name="TRequest"></typeparam>
/// <typeparam name="TResponse"></typeparam>
public class TransactionBehavior<TDbContext, TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse> where TDbContext : EFContext
{
ILogger _logger;
TDbContext _dbContext;
ICapPublisher _capBus;
public TransactionBehavior(TDbContext dbContext, ICapPublisher capBus, ILogger logger)
{
_dbContext = dbContext ?? throw new ArgumentNullException();
_capBus = capBus ?? throw new ArgumentNullException(nameof(capBus));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
/// <summary>
/// 事务执行
/// </summary>
/// <param name="request"></param>
/// <param name="cancellationToken"></param>
/// <param name="next"></param>
/// <returns></returns>
public async Task<TResponse> Handle(TRequest request, CancellationToken cancellationToken, RequestHandlerDelegate<TResponse> next)
{
var response = default(TResponse);
var typeName = request.GetGenericTypeName();
try
{
// 判断当前是否有开启事务，如果开启就执行后续动作
if (_dbContext.HasActiveTransaction)
{
return await next();
}
// 数据库操作默认执行策略
// 比如，可以嵌入重试逻辑
var strategy = _dbContext.Database.CreateExecutionStrategy();
await strategy.ExecuteAsync(async () =>
{
// 开启事务
Guid transactionId;
using (var transaction = await _dbContext.BeginTransactionAsync())
// 记录开启的事务
using (_logger.BeginScope("TransactionContext:{TransactionId}", transaction.TransactionId))
{
_logger.LogInformation("----- 开始事务 {TransactionId} ({@Command})", transaction.TransactionId, typeName, request);
// 类似中间件模式，后续逻辑执行完成后，提交事务
response = await next();
_logger.LogInformation("----- 提交事务 {TransactionId} ({CommandName})", transaction.TransactionId, typeName);
// 提交事务
await _dbContext.CommitTransactionAsync(transaction);
transactionId = transaction.TransactionId;
}
});
return response;
}
catch (Exception ex)
{
_logger.LogError(ex, "处理事务出错 {CommandName} ({@Command})", typeName, request);
throw;
}
}
}
这里可能会有点疑问，这里没有rollback啊。
using (var transaction = await _dbContext.BeginTransactionAsync())
这一句是托管了，如果中间发生异常，那么会自动调用rollback，using原理前面在c# 基础篇中介绍了，本质就是try catch finnaly这样的模式，这里不详细介绍了。
结
下一节仓储层的具体实现。