[ABP]浅谈工作单元 在整个 ABP 框架当中的应用
ABP在其内部实现了工作单元模式,统一地进行事务与连接管理。 其核心就是通过 Castle 的 Dynamic Proxy 进行动态代理,在组件注册的时候进行拦截器注入,拦截到实现了 Unit Of Work 特性的方法进行操作,在执行完方法之后就会关闭掉工作单元。
其整体流程大概如下:
- 首先 UOW 相关接口、拦截器等通过 IocManager 注入到 Ioc 容器当中。
- 监听 Ioc 注册事件,并为其添加方法拦截器。
- 在拦截器内部使用 using 包裹数据库操作方法,使其成为一个工作单元。
-
一旦在方法 procced 执行的时候,产生任何异常触发任何异常都不会执行 Complete 方法,直接抛出终止执行。
UnitOfWorkInterceptors
这是一个 Castle Interceptors 的实现,在 AbpBootStrap 的 Initialze 方法当中被注入到 Ioc 容器。
private void AddInterceptorRegistrars() { ValidationInterceptorRegistrar.Initialize(IocManager); AuditingInterceptorRegistrar.Initialize(IocManager); EntityHistoryInterceptorRegistrar.Initialize(IocManager); UnitOfWorkRegistrar.Initialize(IocManager); AuthorizationInterceptorRegistrar.Initialize(IocManager); }
之后在 Registrar 内部针对组件注入事件进行绑定:
public static void Initialize(IIocManager iocManager) { iocManager.IocContainer.Kernel.ComponentRegistered += (key, handler) => { var implementationType = handler.ComponentModel.Implementation.GetTypeInfo(); HandleTypesWithUnitOfWorkAttribute(implementationType, handler); HandleConventionalUnitOfWorkTypes(iocManager, implementationType, handler); }; }
在这里的两个方法分别是针对已经实现了 UnitOfWork 特性的类进行绑定,另外一个则是针对符合命名规则的类型进行拦截器绑定。
拦截器做的事情十分简单,针对拦截到的方法进行 UOW 特性检测,如果检测通过之后则直接执行工作单元方法,并根据特性生成 Options。不过不是一个 UOW 的话,则直接继续执行原先方法内代码。var unitOfWorkAttr = _unitOfWorkOptions.GetUnitOfWorkAttributeOrNull(method); if (unitOfWorkAttr == null || unitOfWorkAttr.IsDisabled) { //No need to a uow invocation.Proceed(); return; } PerformUow(invocation, unitOfWorkAttr.CreateOptions());
在创建 UOW 的时候,拦截器也会根据方法类型的不同来用不同的方式构建 UOW 。
如果是同步方法的话:private void PerformSyncUow(IInvocation invocation, UnitOfWorkOptions options) { // 直接从 Manager 生成一个新的工作单元 using (var uow = _unitOfWorkManager.Begin(options)) { // 继续执行原方法 // 产生异常直接进入 uow 的 Dispose 方法 invocation.Proceed(); // 如果一切正常,提交事务 uow.Complete(); } }
但如果这个工作单元被标注在异步方法上面,则操作略微复杂:
private void PerformAsyncUow(IInvocation invocation, UnitOfWorkOptions options) { // 获得一个工作单元 var uow = _unitOfWorkManager.Begin(options); // 继续执行拦截到的方法 try { invocation.Proceed(); } catch { uow.Dispose(); throw; } // 如果是异步无返回值的方法 if (invocation.Method.ReturnType == typeof(Task)) { invocation.ReturnValue = InternalAsyncHelper.AwaitTaskWithPostActionAndFinally( (Task) invocation.ReturnValue, async () => await uow.CompleteAsync(), exception => uow.Dispose() ); } else //Task<TResult> { invocation.ReturnValue = InternalAsyncHelper.CallAwaitTaskWithPostActionAndFinallyAndGetResult( invocation.Method.ReturnType.GenericTypeArguments[0], invocation.ReturnValue, async () => await uow.CompleteAsync(), exception => uow.Dispose() ); } }
以上代码在进入的时候直接执行原方法,如果产生任何异常直接进入 Dispose 方法并且抛出异常。乍一看与同步方法处理没什么区别,但重点是这里并没有执行 Complete 方法,因为这里需要对其异步返回结果更改为 UOW 异步提交之后的值,先查看第一种直接返回 Task 的情况.
public static async Task AwaitTaskWithPostActionAndFinally(Task actualReturnValue, Func<Task> postAction, Action<Exception> finalAction) { Exception exception = null; try { // 原有异步任务返回值 await actualReturnValue; // 新的异步返回结果 await postAction(); } catch (Exception ex) { exception = ex; throw; } finally { finalAction(exception); } }
在内部首先等待原有任务执行完成之后再执行传入的 UOW 的 CompleteAsync() 方法,并且在执行过程中有无异常都会直接调用 UOW 的 Disopse 释放资源。
第二种则适用于有返回值的情况:public static object CallAwaitTaskWithPostActionAndFinallyAndGetResult(Type taskReturnType, object actualReturnValue, Func<Task> action, Action<Exception> finalAction) { // 获得 AwaitTaskWithPreActionAndPostActionAndFinallyAndGetResult 方法,并且通过反射构造一个泛型方法,并且将自身参数传入调用。 return typeof (InternalAsyncHelper) .GetMethod("AwaitTaskWithPostActionAndFinallyAndGetResult", BindingFlags.Public | BindingFlags.Static) .MakeGenericMethod(taskReturnType) .Invoke(null, new object[] { actualReturnValue, action, finalAction }); } public static async Task<T> AwaitTaskWithPreActionAndPostActionAndFinallyAndGetResult<T>(Func<Task<T>> actualReturnValue, Func<Task> preAction = null, Func<Task> postAction = null, Action<Exception> finalAction = null) { Exception exception = null; try { if (preAction != null) { await preAction(); } var result = await actualReturnValue(); if (postAction != null) { await postAction(); } return result; } catch (Exception ex) { exception = ex; throw; } finally { if (finalAction != null) { finalAction(exception); } } }
这两个方法的作用都是确保 CompleteAsync 和 Dispose 能够在放在异步任务当中执行。
IUnitOfWorkManager
顾名思义,这是一个 UOW 的管理器,在 ABP 内部有其一个默认实现
UnitOfWorkManager
,在 AbpKernel 模块初始化的时候就已经被注入。
核心方法是 Begin 方法,在 Begin 方法当中通过FillDefaultsForNonProvidedOptions
方法判断是否传入了配置参数,如果没有传入的话则构建一个默认参数对象。public IUnitOfWorkCompleteHandle Begin(UnitOfWorkOptions options) { // 如果没有 UOW 参数,构造默认参数 options.FillDefaultsForNonProvidedOptions(_defaultOptions); var outerUow = _currentUnitOfWorkProvider.Current; // 当前是否已经存在工作单元,存在工作单元的话,构建一个内部工作单元 if (options.Scope == TransactionScopeOption.Required && outerUow != null) { return new InnerUnitOfWorkCompleteHandle(); } // 不存在的话构建一个新的工作单元 var uow = _iocResolver.Resolve<IUnitOfWork>(); // 绑定各种事件 uow.Completed += (sender, args) => { _currentUnitOfWorkProvider.Current = null; }; uow.Failed += (sender, args) => { _currentUnitOfWorkProvider.Current = null; }; uow.Disposed += (sender, args) => { _iocResolver.Release(uow); }; //Inherit filters from outer UOW if (outerUow != null) { options.FillOuterUowFiltersForNonProvidedOptions(outerUow.Filters.ToList()); } uow.Begin(options); //Inherit tenant from outer UOW if (outerUow != null) { uow.SetTenantId(outerUow.GetTenantId(), false); } // 设置当前工作单元为新创建的 UOW _currentUnitOfWorkProvider.Current = uow; return uow; }
这里 Begin 方法所返回的是一个
IUnitOfWorkCompleteHandle
对象,跳转到IUnitOfWorkCompleteHandle
,可以看到它有两个方法:public interface IUnitOfWorkCompleteHandle : IDisposable { void Complete(); Task CompleteAsync(); }
都是完成工作单元的方法,一个同步、一个异步,同时这个接口也实现了 IDispose 接口,从开始使用 using 就可以看出其含义。
InnerUnitOfWorkCompleteHandle
参考其引用,可以发现有一个
IUnitOfWork
接口继承自它,并且还有一个InnerUnitOfWorkCompleteHandle
的内部实现。这里先查看InnerUnitOfWorkCompleteHandle
内部实现:internal class InnerUnitOfWorkCompleteHandle : IUnitOfWorkCompleteHandle { public const string DidNotCallCompleteMethodExceptionMessage = "Did not call Complete method of a unit of work."; private volatile bool _isCompleteCalled; private volatile bool _isDisposed; public void Complete() { _isCompleteCalled = true; } public Task CompleteAsync() { _isCompleteCalled = true; return Task.FromResult(0); } public void Dispose() { if (_isDisposed) { return; } _isDisposed = true; if (!_isCompleteCalled) { if (HasException()) { return; } throw new AbpException(DidNotCallCompleteMethodExceptionMessage); } } private static bool HasException() { try { return Marshal.GetExceptionCode() != 0; } catch (Exception) { return false; } } }
其内部实现十分简单,其中 Complete 的同步和异步方法都只是对完成标识进行一个标记。并未真正的进行任何数据库事务操作。同时在他的内部也实现了
IDispose
接口,如果 complete 未被标记为已完成,那么直接抛出异常,后续操作不会执行。
现在再转到 Begin 方法内部就可以发现,在创建的时候会首先判断了当前是否已经存在了工作单元,如果存在了才会创建这样一个内部工作单元。也就是说真正那个的事务操作是在返回的 IUnitOfWork 当中实现的。
这样的话就会构建出一个嵌套的工作单元:OuterUOW->InnerUow1->InnerUOW2->.....->InnerUowN
你可以想象有以下代码:
public void TestUowMethod() { using(var outerUOW = Manager.Begin()) // 这里返回的是 IOC 解析出的 IUnitOfWork { OperationOuter(); using(var innerUOW1 = Manager.Begin()) // 内部 UOW { Operation1(); using(var innerUOW2 = Manager.Begin()) // 内部 UOW { Operation2(); Complete(); } Complete(); } Complete(); } }
当代码执行的时候,如同俄罗斯套娃,从内部依次到外部执行,内部工作单元仅会在调用 Complete 方法的时候将 completed 标记为 true,但一旦操作抛出异常,Complete无法得到执行,则会直接抛出异常,中断外层代码执行。
那么 IUnitOfWork 的实现又是怎样的呢?
在 ABP 内部针对 EF Core 框架实现了一套 UOW,其继承自 UnitOfWorkBase,而在 UnitOfWorkBase 内部有部分针对接口 IActiveUnitOfWork 的实现,同时由于 IUnifOfWork 也实现了 IUnitOfWorkCompleteHandle 接口,所以在 Begin 方法处能够向上转型。IActiveUnitOfWork
在 IActiveUnitOfWork 接口当中定义了工作单元的三个事件,用户在使用的时候可以针对这三个事件进行绑定。
| 事件名称 | 触发条件 |
| ------- | --------- |
| Completed | 工作单元调用 Complete 方法之后触发 |
| Failed | 工作单元在调用 Complete 方法如果产生异常,则会在 Dispose 释放资源时触发。 |
| Disposed | 释放工作单元的时候触发 |
除了三个事件之外,在 ABP 当中的数据过滤器也是在工作单元调用的时候工作的,后面讲解 EfCoreUnitOfWork 的时候会研究,数据过滤仅包括软删除等审计字段,同时也包括多租户的租户 ID 也在工作单元当中进行设置的。
在这里也定义了两个 SaveChanges 方法,一个同步、一个异步,作用跟实体上下文的同名方法作用差不多。IUnitOfWork
IUnitOfWork 同时继承了 IUnitOfWorkCompleteHandle 与 IActiveUnitOfWork 接口,并且增加了两个属性与一个 Begin 方法。
UnitOfWorkBase
UnitOfWorkBase 是所有工作单元的一个抽象基类,在其内部实现了 IUnitOfWork 的所有方法,其中也包括两个父级接口。
下面就一个一个属性讲解。Id
这个 Id 是在构造函数当中初始化的,全局唯一的 Id,使用 Guid 初始化其值。
Outer
用于标明当前工作单元的外部工作单元,其值是在 UnitOfWorkManager 创建工作单元的时候赋予的。在 Manager 的 Begin 方法当中每次针对 Current Uow 赋值的时候都会将已经存在的 UOW 关联最新的 Current Uow 的 Outer 属性上面。形成如下结构:
Current Uow = Uow3.Outer = Uow2.Outer = Uow1.Outer = null
具体代码参考
ICurrentUnitOfWorkProvider
的实现。Begin()
在 Manager 创建好一个 Uow 之后,就会调用它的 Begin 方法,在内部首先做了一个判断,判断是否多次调用了 Begin 方法,这里可以看到并未做加锁处理。这是因为在 ABP 当中,一个线程当*用一个工作单元。其实在 CurrentProvider 的代码当中就可以看出来如果在一个线程里面创建了多个工作单元,他会将其串联起来。
之后设置过滤器,并且开始调用 BeginUow 方法,这里并未实现,具体实现我们转到 EfUnitOfWork 可以看到。BeginUow()-EfCoreUnitOfWork
protected override void BeginUow() { if (Options.IsTransactional == true) { _transactionStrategy.InitOptions(Options); } }
覆写了父类方法,仅对设置进行初始化。
其实到这里我们就大概清楚了 ABP 整个 UOW 的工作流程,如果是两个打上 UnitOfWork 特性的方法在 A 调用 B 的时候其实就会封装成两个嵌套的 using 块。using(var a = Manager.Begin()) { // 操作 using(var b = Manager.Begin()) { // 操作 b.Complete(); } a.Complete(); }
而最外层的 Complete 就是真正执行了数据库事务操作的。
可以看下 EfUnitOfWork 的实现:// 这里是 UnitOfWorkBase 的 Complete 方法 public void Complete() { PreventMultipleComplete(); try { CompleteUow(); _succeed = true; OnCompleted(); } catch (Exception ex) { _exception = ex; throw; } }
// 这里是 EfUnitOfWork 的 CompleteUow 方法 protected override void CompleteUow() { SaveChanges(); CommitTransaction(); } // 遍历所有激活的 DbContext 并保存更改到数据库 public override void SaveChanges() { foreach (var dbContext in GetAllActiveDbContexts()) { SaveChangesInDbContext(dbContext); } } // 提交事务 private void CommitTransaction() { if (Options.IsTransactional == true) { _transactionStrategy.Commit(); } }