fix: 解决dotnetcore.cap 事务一致性问题

aspnet-core/Lion.AbpPro.sln

@@ -240,6 +240,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Lion.AbpPro.LanguageManagem
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Lion.AbpPro.LanguageManagement.Application.Tests", "modules\LanguageManagement\test\Lion.AbpPro.LanguageManagement.Application.Tests\Lion.AbpPro.LanguageManagement.Application.Tests.csproj", "{15852D6F-4110-4B98-B89D-5747777E8908}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Lion.AbpPro.CAP.EntityFrameworkCore", "frameworks\src\Lion.AbpPro.CAP.EntityFrameworkCore\Lion.AbpPro.CAP.EntityFrameworkCore.csproj", "{68C902A2-A604-4F3A-879D-37941C00C7A9}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -582,6 +584,10 @@ Global
 		{15852D6F-4110-4B98-B89D-5747777E8908}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{15852D6F-4110-4B98-B89D-5747777E8908}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{15852D6F-4110-4B98-B89D-5747777E8908}.Release|Any CPU.Build.0 = Release|Any CPU
+		{68C902A2-A604-4F3A-879D-37941C00C7A9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{68C902A2-A604-4F3A-879D-37941C00C7A9}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{68C902A2-A604-4F3A-879D-37941C00C7A9}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{68C902A2-A604-4F3A-879D-37941C00C7A9}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -696,6 +702,7 @@ Global
 		{8872921C-5BF9-4B4C-B882-6AF8CC78D2CF} = {3FE23400-A323-46ED-A7F4-30BFF8916D64}
 		{E5994C85-C1C2-44F3-BF10-3277CA6CF2C9} = {3FE23400-A323-46ED-A7F4-30BFF8916D64}
 		{15852D6F-4110-4B98-B89D-5747777E8908} = {3FE23400-A323-46ED-A7F4-30BFF8916D64}
+		{68C902A2-A604-4F3A-879D-37941C00C7A9} = {7BE85EBC-99AD-4CDE-957E-4BDD087FC4E3}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {28315BFD-90E7-4E14-A2EA-F3D23AF4126F}

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/GlobalUsings.cs

@@ -0,0 +1,13 @@
+// Global using directives
+
+global using System.Collections.Concurrent;
+global using DotNetCore.CAP;
+global using Microsoft.EntityFrameworkCore;
+global using Microsoft.EntityFrameworkCore.Storage;
+global using Microsoft.Extensions.DependencyInjection;
+global using Microsoft.Extensions.Options;
+global using Volo.Abp.DependencyInjection;
+global using Volo.Abp.Modularity;
+global using Volo.Abp.Threading;
+global using Volo.Abp.Uow;
+global using Volo.Abp.Uow.EntityFrameworkCore;

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion.AbpPro.CAP.EntityFrameworkCore.csproj

@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net7.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <RootNamespace />
+        <AssemblyName>Lion.AbpPro.CAP.EntityFrameworkCore</AssemblyName>
+    </PropertyGroup>
+    <ItemGroup>
+        <PackageReference Include="Volo.Abp.EntityFrameworkCore" />
+    </ItemGroup>
+    <ItemGroup>
+      <ProjectReference Include="..\Lion.AbpPro.CAP\Lion.AbpPro.CAP.csproj" />
+    </ItemGroup>
+</Project>

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/EfCoreLionAbpProCapTransactionApiFactory.cs

@@ -0,0 +1,64 @@
+namespace Lion.AbpPro.CAP.EntityFrameworkCore;
+
+public class EfCoreLionAbpProCapTransactionApiFactory : ILionAbpProCapTransactionApiFactory, ITransientDependency
+{
+    public Type TransactionApiType { get; } = typeof(EfCoreTransactionApi);
+
+    protected readonly ICapPublisher Publisher;
+    protected readonly LionAbpProEfCoreDbContextCapOptions Options;
+    protected readonly ILionAbpProCapDbProviderInfoProvider LionAbpProCapDbProviderInfoProvider;
+    protected readonly ICancellationTokenProvider CancellationTokenProvider;
+
+    public EfCoreLionAbpProCapTransactionApiFactory(
+        ICapPublisher publisher,
+        IOptions<LionAbpProEfCoreDbContextCapOptions> options,
+        ILionAbpProCapDbProviderInfoProvider lionAbpProCapDbProviderInfoProvider,
+        ICancellationTokenProvider cancellationTokenProvider)
+    {
+        Publisher = publisher;
+        Options = options.Value;
+        LionAbpProCapDbProviderInfoProvider = lionAbpProCapDbProviderInfoProvider;
+        CancellationTokenProvider = cancellationTokenProvider;
+    }
+
+    public virtual ITransactionApi Create(ITransactionApi originalApi)
+    {
+        var efApi = (EfCoreTransactionApi)originalApi;
+
+        var capTrans = CreateCapTransactionOrNull(efApi);
+
+        return capTrans is null
+            ? originalApi
+            : new EfCoreTransactionApi(capTrans, efApi.StarterDbContext, CancellationTokenProvider);
+    }
+
+    protected virtual IDbContextTransaction CreateCapTransactionOrNull(EfCoreTransactionApi originalApi)
+    {
+        // TODO 通过数据库连接字符串判断数据库类型
+        // if (Options.CapUsingDbConnectionString != originalApi.StarterDbContext.Database.GetConnectionString())
+        // {
+        //     return null;
+        // }
+
+        var dbProviderInfo = LionAbpProCapDbProviderInfoProvider.GetOrNull(originalApi.StarterDbContext.Database.ProviderName);
+
+        if (dbProviderInfo?.CapTransactionType is null || dbProviderInfo.CapEfDbTransactionType is null)
+        {
+            return null;
+        }
+
+        var capTransactionType = dbProviderInfo.CapTransactionType;
+
+        if (ActivatorUtilities.CreateInstance(Publisher.ServiceProvider, capTransactionType) is not CapTransactionBase capTransaction)
+        {
+            return null;
+        }
+
+        capTransaction.DbTransaction = originalApi.DbContextTransaction;
+        capTransaction.AutoCommit = false;
+
+        Publisher.Transaction.Value = capTransaction;
+
+        return (IDbContextTransaction)Activator.CreateInstance(dbProviderInfo.CapEfDbTransactionType, capTransaction);
+    }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/ILionAbpProCapDbProviderInfoProvider.cs

@@ -0,0 +1,6 @@
+namespace Lion.AbpPro.CAP.EntityFrameworkCore;
+
+public interface ILionAbpProCapDbProviderInfoProvider
+{
+    LionAbpProCapDbProviderInfo GetOrNull(string dbProviderName);
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/LionAbpProCapDbProviderInfo.cs

@@ -0,0 +1,14 @@
+namespace Lion.AbpPro.CAP.EntityFrameworkCore;
+
+public class LionAbpProCapDbProviderInfo
+{
+    public Type CapTransactionType { get; }
+    
+    public Type CapEfDbTransactionType { get; }
+    
+    public LionAbpProCapDbProviderInfo(string capTransactionTypeName, string capEfDbTransactionTypeName)
+    {
+        CapTransactionType = Type.GetType(capTransactionTypeName, false);
+        CapEfDbTransactionType = Type.GetType(capEfDbTransactionTypeName, false);
+    }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/LionAbpProCapEntityFrameworkCoreModule.cs

@@ -0,0 +1,7 @@
+namespace Lion.AbpPro.CAP.EntityFrameworkCore
+{
+    [DependsOn(typeof(LionAbpProCapModule))]
+    public class LionAbpProCapEntityFrameworkCoreModule : AbpModule
+    {
+    }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/LionAbpProCapOptionsExtensions.cs

@@ -0,0 +1,18 @@
+using Lion.AbpPro.CAP.EntityFrameworkCore;
+
+// ReSharper disable once CheckNamespace
+namespace Microsoft.Extensions.DependencyInjection
+{
+    public static class LionAbpProCapOptionsExtensions
+    {
+        public static CapOptions SetCapDbConnectionString(this CapOptions options, string dbConnectionString)
+        {
+            options.RegisterExtension(new LionAbpProEfCoreDbContextCapOptionsExtension
+            {
+                CapUsingDbConnectionString = dbConnectionString
+            });
+            
+            return options;
+        }
+    }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/LionAbpProEfCoreDbContextCapOptions.cs

@@ -0,0 +1,6 @@
+namespace Lion.AbpPro.CAP.EntityFrameworkCore;
+
+public class LionAbpProEfCoreDbContextCapOptions
+{
+    public string CapUsingDbConnectionString { get; set; }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/LionAbpProEfCoreDbContextCapOptionsExtension.cs

@@ -0,0 +1,14 @@
+namespace Lion.AbpPro.CAP.EntityFrameworkCore;
+
+public class LionAbpProEfCoreDbContextCapOptionsExtension : ICapOptionsExtension
+{
+    public string CapUsingDbConnectionString { get; init; }
+    
+    public void AddServices(IServiceCollection services)
+    {
+        services.Configure<LionAbpProEfCoreDbContextCapOptions>(options =>
+        {
+            options.CapUsingDbConnectionString = CapUsingDbConnectionString;
+        });
+    }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/LionAbpProLionAbpProCapDbProviderInfoProvider.cs

@@ -0,0 +1,45 @@
+namespace Lion.AbpPro.CAP.EntityFrameworkCore;
+
+public class LionAbpProLionAbpProCapDbProviderInfoProvider : ILionAbpProCapDbProviderInfoProvider, ITransientDependency
+{
+    protected ConcurrentDictionary<string, LionAbpProCapDbProviderInfo> CapDbProviderInfos { get; set; } = new();
+
+    public virtual LionAbpProCapDbProviderInfo GetOrNull(string dbProviderName)
+    {
+        return CapDbProviderInfos.GetOrAdd(dbProviderName, InternalGetOrNull);
+    }
+    
+    protected virtual LionAbpProCapDbProviderInfo InternalGetOrNull(string databaseProviderName)
+    {
+        switch (databaseProviderName)
+        {
+            case "Microsoft.EntityFrameworkCore.SqlServer":
+                return new LionAbpProCapDbProviderInfo(
+                    "DotNetCore.CAP.SqlServerCapTransaction, DotNetCore.CAP.SqlServer",
+                    "Microsoft.EntityFrameworkCore.Storage.CapEFDbTransaction, DotNetCore.CAP.SqlServer");
+            case "Npgsql.EntityFrameworkCore.PostgreSQL":
+                return new LionAbpProCapDbProviderInfo(
+                    "DotNetCore.CAP.PostgreSqlCapTransaction, DotNetCore.CAP.PostgreSql",
+                    "Microsoft.EntityFrameworkCore.Storage.CapEFDbTransaction, DotNetCore.CAP.PostgreSQL");
+            case "Pomelo.EntityFrameworkCore.MySql":
+                return new LionAbpProCapDbProviderInfo(
+                    "DotNetCore.CAP.MySqlCapTransaction, DotNetCore.CAP.MySql",
+                    "Microsoft.EntityFrameworkCore.Storage.CapEFDbTransaction, DotNetCore.CAP.MySql");
+            case "Oracle.EntityFrameworkCore":
+            case "Devart.Data.Oracle.Entity.EFCore":
+                return new LionAbpProCapDbProviderInfo(
+                    "DotNetCore.CAP.OracleCapTransaction, DotNetCore.CAP.Oracle",
+                    "Microsoft.EntityFrameworkCore.Storage.CapEFDbTransaction, DotNetCore.CAP.Oracle");
+            case "Microsoft.EntityFrameworkCore.Sqlite":
+                return new LionAbpProCapDbProviderInfo(
+                    "DotNetCore.CAP.SqliteCapTransaction, DotNetCore.CAP.Sqlite",
+                    "Microsoft.EntityFrameworkCore.Storage.CapEFDbTransaction, DotNetCore.CAP.Sqlite");
+            case "Microsoft.EntityFrameworkCore.InMemory":
+                return new LionAbpProCapDbProviderInfo(
+                    "DotNetCore.CAP.InMemoryCapTransaction, DotNetCore.CAP.InMemoryStorage",
+                    "Microsoft.EntityFrameworkCore.Storage.CapEFDbTransaction, DotNetCore.CAP.InMemoryStorage");
+            default:
+                return null;
+        }
+    }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/ILionAbpProCapTransactionApiFactory.cs

@@ -0,0 +1,8 @@
+namespace Lion.AbpPro.CAP;
+
+public interface ILionAbpProCapTransactionApiFactory
+{
+    Type TransactionApiType { get; }
+    
+    ITransactionApi Create(ITransactionApi originalApi);
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/LionAbpProCapConsumerServiceSelector.cs

@@ -1,19 +1,14 @@
 namespace Lion.AbpPro.CAP;
 
-[Dependency(ServiceLifetime.Singleton, ReplaceServices = true)]
+public class LionAbpProCapConsumerServiceSelector : ConsumerServiceSelector, ISingletonDependency
-[ExposeServices(typeof(IConsumerServiceSelector))]
-public sealed class LionAbpProCapConsumerServiceSelector : ConsumerServiceSelector
 {
-    private AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; }
+    protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; }
-    private IServiceProvider ServiceProvider { get; }
+    protected IServiceProvider ServiceProvider { get; }
 
     /// <summary>
     /// Creates a new <see cref="T:DotNetCore.CAP.Internal.ConsumerServiceSelector" />.
     /// </summary>
-    public LionAbpProCapConsumerServiceSelector(
+    public LionAbpProCapConsumerServiceSelector(IServiceProvider serviceProvider, IOptions<AbpDistributedEventBusOptions> distributedEventBusOptions) : base(serviceProvider)
-        IServiceProvider serviceProvider, 
-        IOptions<AbpDistributedEventBusOptions> distributedEventBusOptions) 
-        : base(serviceProvider)
     {
         ServiceProvider = serviceProvider;
         AbpDistributedEventBusOptions = distributedEventBusOptions.Value;
@@ -22,7 +17,7 @@ public sealed class LionAbpProCapConsumerServiceSelector : ConsumerServiceSelect
     protected override IEnumerable<ConsumerExecutorDescriptor> FindConsumersFromInterfaceTypes(IServiceProvider provider)
     {
         var executorDescriptorList = base.FindConsumersFromInterfaceTypes(provider).ToList();
-            
+
         //handlers
         var handlers = AbpDistributedEventBusOptions.Handlers;
 
@@ -35,8 +30,9 @@ public sealed class LionAbpProCapConsumerServiceSelector : ConsumerServiceSelect
                 {
                     continue;
                 }
+
                 var genericArgs = @interface.GetGenericArguments();
-                    
+
                 if (genericArgs.Length != 1)
                 {
                     continue;
@@ -51,17 +47,18 @@ public sealed class LionAbpProCapConsumerServiceSelector : ConsumerServiceSelect
 
                     descriptor.Attribute.Group = descriptor.Attribute.Group.Insert(
                         descriptor.Attribute.Group.LastIndexOf(".", StringComparison.Ordinal), $".{count}");
-                            
+
                     executorDescriptorList.Add(descriptor);
                 }
-                        
+
                 //Subscribe(genericArgs[0], new IocEventHandlerFactory(ServiceScopeFactory, handler));
             }
         }
+
         return executorDescriptorList;
     }
 
-    private IEnumerable<ConsumerExecutorDescriptor> GetHandlerDescription(Type eventType,Type typeInfo)
+    protected virtual IEnumerable<ConsumerExecutorDescriptor> GetHandlerDescription(Type eventType, Type typeInfo)
     {
         var serviceTypeInfo = typeof(IDistributedEventHandler<>)
             .MakeGenericType(eventType);

aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/LionAbpProCapDistributedEventBus.cs

@@ -1,24 +1,21 @@
 namespace Lion.AbpPro.CAP;
 
-public class LionAbpProCapDistributedEventBus :
+public class LionAbpProCapDistributedEventBus : EventBusBase, IDistributedEventBus, ISingletonDependency
-    EventBusBase,
-    IDistributedEventBus,
-    ISingletonDependency
 {
-    private AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; }
+    protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; }
-    private ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; }
+    protected readonly ICapPublisher CapPublisher;
-    private ConcurrentDictionary<string, Type> EventTypes { get; }
 
-    private readonly ICapPublisher CapPublisher;
+    //TODO: Accessing to the List<IEventHandlerFactory> may not be thread-safe!
+    protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; }
+    protected ConcurrentDictionary<string, Type> EventTypes { get; }
 
-        
     public LionAbpProCapDistributedEventBus(IServiceScopeFactory serviceScopeFactory,
         IOptions<AbpDistributedEventBusOptions> distributedEventBusOptions,
         ICapPublisher capPublisher,
+        IUnitOfWorkManager unitOfWorkManager,
         ICurrentTenant currentTenant,
-        UnitOfWorkManager unitOfWorkManager,
         IEventHandlerInvoker eventHandlerInvoker)
-        : base(serviceScopeFactory, currentTenant,unitOfWorkManager,eventHandlerInvoker)
+        : base(serviceScopeFactory, currentTenant, unitOfWorkManager, eventHandlerInvoker)
     {
         CapPublisher = capPublisher;
         AbpDistributedEventBusOptions = distributedEventBusOptions.Value;
@@ -82,36 +79,63 @@ public class LionAbpProCapDistributedEventBus :
         GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear());
     }
 
-    protected override Task PublishToEventBusAsync(Type eventType, object eventData)
+    public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class
     {
-        throw new NotImplementedException();
+        return Subscribe(typeof(TEvent), handler);
     }
 
-    protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord)
+    public virtual Task PublishAsync<TEvent>(TEvent eventData, bool onUnitOfWorkComplete = true,
+        bool useOutbox = true) where TEvent : class
     {
-        throw new NotImplementedException();
+        return PublishAsync(typeof(TEvent), eventData, onUnitOfWorkComplete, useOutbox);
     }
 
-    public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class
+    public virtual async Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true,
+        bool useOutbox = true)
     {
-        return Subscribe(typeof(TEvent), handler);
+        if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null)
-    }
+        {
+            AddToUnitOfWork(
+                UnitOfWorkManager.Current,
+                new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext(), useOutbox)
+            );
+            return;
+        }
 
-    public async Task PublishAsync<TEvent>(TEvent eventData, bool onUnitOfWorkComplete = true,
+        if (useOutbox && UnitOfWorkManager.Current != null)
-        bool useOutbox = true) where TEvent : class
+        {
-    {
+            if (UnitOfWorkManager.Current is not LionAbpProCapUnitOfWork capUnitOfWork || capUnitOfWork.CapTransaction is null)
-        var eventName = EventNameAttribute.GetNameOrDefault(typeof(TEvent));
+            {
-        await CapPublisher.PublishAsync(eventName, eventData);
+                UnitOfWorkManager.Current.OnCompleted(async () =>
+                {
+                    await PublishToEventBusAsync(eventType, eventData);
+                });
+            }
+            else
+            {
+                using (CapPublisher.UseTransaction(capUnitOfWork.CapTransaction))
+                {
+                    // Use CAP transactional outbox
+                    await PublishToEventBusAsync(eventType, eventData);
+                }
+            }
 
+            return;
+        }
+
+        await PublishToEventBusAsync(eventType, eventData);
     }
 
-    public async Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true,
+    protected override async Task PublishToEventBusAsync(Type eventType, object eventData)
-        bool useOutbox = true)
     {
         var eventName = EventNameAttribute.GetNameOrDefault(eventType);
         await CapPublisher.PublishAsync(eventName, eventData);
     }
-        
+
+    protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord)
+    {
+        unitOfWork.AddOrReplaceDistributedEvent(eventRecord);
+    }
 
     protected override IEnumerable<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType)
     {

aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/LionAbpProCapModule.cs

@@ -2,7 +2,8 @@ namespace Lion.AbpPro.CAP;
 
 [DependsOn(
     typeof(AbpEventBusModule), 
-    typeof(LionAbpProLocalizationModule))]
+    typeof(LionAbpProLocalizationModule),
+    typeof(AbpUnitOfWorkModule))]
 public class LionAbpProCapModule : AbpModule
 {
 }

aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/LionAbpProCapPublisherExtension.cs

@@ -0,0 +1,11 @@
+namespace Lion.AbpPro.CAP;
+
+public static class LionAbpProCapPublisherExtension
+{
+    public static IDisposable UseTransaction(this ICapPublisher capPublisher, ICapTransaction capTransaction)
+    {
+        var previousValue = capPublisher.Transaction.Value;
+        capPublisher.Transaction.Value = capTransaction;
+        return new DisposeAction(() => capPublisher.Transaction.Value = previousValue);
+    }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/LionAbpProCapServiceCollectionExtensions.cs

@@ -1,3 +1,6 @@
+using DotNetCore.CAP.Serialization;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+
 namespace Lion.AbpPro.CAP;
 
 public static class LionAbpProCapServiceCollectionExtensions
@@ -6,9 +9,15 @@ public static class LionAbpProCapServiceCollectionExtensions
         this ServiceConfigurationContext context, 
         Action<CapOptions> capAction)
     {
+        // context.Services.AddSingleton<IConsumerServiceSelector, LionAbpProCapConsumerServiceSelector>();
+        // context.Services.AddSingleton<IDistributedEventBus, LionAbpProCapDistributedEventBus>();
+        // context.Services.AddSingleton<ISerializer, LionAbpProJsonSerializer>();
+        
+        context.Services.Replace(ServiceDescriptor.Transient<IUnitOfWork, LionAbpProCapUnitOfWork>());
+        context.Services.Replace(ServiceDescriptor.Transient<UnitOfWork, LionAbpProCapUnitOfWork>());
+        context.Services.AddTransient<LionAbpProCapUnitOfWork>();
+        
         context.Services.AddCap(capAction);
-        context.Services.AddSingleton<IConsumerServiceSelector, LionAbpProCapConsumerServiceSelector>();
-        context.Services.AddSingleton<IDistributedEventBus, LionAbpProCapDistributedEventBus>();
         return context;
     }
 }

aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/LionAbpProCapUnitOfWork.cs

@@ -0,0 +1,51 @@
+namespace Lion.AbpPro.CAP;
+
+[Dependency(TryRegister = true)]
+public class LionAbpProCapUnitOfWork : UnitOfWork
+{
+    public ICapTransaction CapTransaction { get; protected set; }
+
+    protected ICapPublisher CapPublisher { get; }
+
+    public LionAbpProCapUnitOfWork(
+        IServiceProvider serviceProvider,
+        IUnitOfWorkEventPublisher unitOfWorkEventPublisher,
+        IOptions<AbpUnitOfWorkDefaultOptions> options,
+        ICapPublisher capPublisher)
+        : base(serviceProvider, unitOfWorkEventPublisher, options)
+    {
+        CapPublisher = capPublisher;
+    }
+
+    public override void AddTransactionApi(string key, ITransactionApi api)
+    {
+        var factories = ServiceProvider.GetServices<ILionAbpProCapTransactionApiFactory>();
+
+        var factory = factories.FirstOrDefault(x => x.TransactionApiType == api.GetType());
+
+        if (factory is not null)
+        {
+            api = factory.Create(api);
+            CapTransaction = CapPublisher.Transaction.Value;
+        }
+        
+        base.AddTransactionApi(key, api);
+    }
+
+    public override ITransactionApi GetOrAddTransactionApi(string key, Func<ITransactionApi> factory)
+    {
+        Check.NotNull(key, nameof(key));
+        Check.NotNull(factory, nameof(factory));
+
+        var transactionApi = FindTransactionApi(key);
+
+        if (transactionApi is not null)
+        {
+            return transactionApi;
+        }
+        
+        AddTransactionApi(key, factory());
+
+        return FindTransactionApi(key);
+    }
+}

aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/LionAbpProJsonSerializer.cs

@@ -0,0 +1,67 @@
+using System.Text;
+using System.Text.Json;
+using DotNetCore.CAP.Messages;
+using DotNetCore.CAP.Serialization;
+using Volo.Abp.Json;
+
+namespace Lion.AbpPro.CAP;
+
+public class LionAbpProJsonSerializer : ISerializer, ISingletonDependency
+{
+    private readonly IJsonSerializer _jsonSerializer;
+
+    public LionAbpProJsonSerializer(IJsonSerializer jsonSerializer)
+    {
+        _jsonSerializer = jsonSerializer;
+    }
+
+    public virtual string Serialize(Message message)
+    {
+        return _jsonSerializer.Serialize(message);
+    }
+
+    public virtual ValueTask<TransportMessage> SerializeAsync(Message message)
+    {
+        if (message == null)
+        {
+            throw new ArgumentNullException(nameof(message));
+        }
+
+        if (message.Value == null)
+        {
+            return new ValueTask<TransportMessage>(new TransportMessage(message.Headers, null));
+        }
+
+        var json = _jsonSerializer.Serialize(message.Value);
+
+        return new ValueTask<TransportMessage>(new TransportMessage(message.Headers, Encoding.UTF8.GetBytes(json)));
+    }
+
+    public virtual Message Deserialize(string json)
+    {
+        return _jsonSerializer.Deserialize<Message>(json);
+    }
+
+    public virtual ValueTask<Message> DeserializeAsync(TransportMessage transportMessage, Type valueType)
+    {
+        if (valueType == null || transportMessage.Body.IsEmpty)
+        {
+            return new ValueTask<Message>(new Message(transportMessage.Headers, null));
+        }
+
+        var json = Encoding.UTF8.GetString(transportMessage.Body.ToArray());
+
+        return new ValueTask<Message>(new Message(transportMessage.Headers,
+            _jsonSerializer.Deserialize(valueType, json)));
+    }
+
+    public virtual object Deserialize(object value, Type valueType)
+    {
+        return _jsonSerializer.Deserialize(valueType, value.ToString());
+    }
+
+    public virtual bool IsJsonType(object jsonObject)
+    {
+        return jsonObject is JsonElement;
+    }
+}

aspnet-core/modules/BasicManagement/src/Lion.AbpPro.BasicManagement.Application/Users/UserAppService.cs

@@ -4,6 +4,7 @@ using Magicodes.ExporterAndImporter.Excel.AspNetCore;
 using Microsoft.AspNetCore.Identity;
 using Microsoft.AspNetCore.Mvc;
 using Volo.Abp.Account;
+using Volo.Abp.Uow;
 using IdentityRole = Volo.Abp.Identity.IdentityRole;
 
 namespace Lion.AbpPro.BasicManagement.Users
@@ -80,7 +81,12 @@ namespace Lion.AbpPro.BasicManagement.Users
         {
             // abp 5.0 之后新增字段,是否运行用户登录，默认设置为true
             input.IsActive = true;
-            return await _identityUserAppService.CreateAsync(input);
+            using (var uow = UnitOfWorkManager.Begin(new AbpUnitOfWorkOptions() { IsTransactional = true }, true))
+            {
+                await _identityUserAppService.CreateAsync(input);
+                await uow.CompleteAsync();
+            }
+            return null;
         }
 
         /// <summary>
@@ -90,7 +96,12 @@ namespace Lion.AbpPro.BasicManagement.Users
         public virtual async Task<IdentityUserDto> UpdateAsync(UpdateUserInput input)
         {
             input.UserInfo.IsActive = true;
-            return await _identityUserAppService.UpdateAsync(input.UserId, input.UserInfo);
+            using (var uow = UnitOfWorkManager.Begin(new AbpUnitOfWorkOptions() { IsTransactional = true }, true))
+            {
+                await _identityUserAppService.UpdateAsync(input.UserId, input.UserInfo);
+            }
+
+            return null;
         }
 
         /// <summary>

aspnet-core/services/host/Lion.AbpPro.HttpApi.Host/AbpProHttpApiHostModule.cs

@@ -1,4 +1,5 @@
 using Hangfire.Redis;
+using Lion.AbpPro.CAP.EntityFrameworkCore;
 using Swagger;
 using Volo.Abp.BackgroundJobs.Hangfire;
 using Volo.Abp.Timing;
@@ -15,6 +16,7 @@ namespace Lion.AbpPro
         typeof(AbpAccountWebModule),
         typeof(AbpProApplicationModule),
         typeof(LionAbpProCapModule),
+        typeof(LionAbpProCapEntityFrameworkCoreModule),
         typeof(AbpAspNetCoreMvcUiBasicThemeModule),
         typeof(AbpCachingStackExchangeRedisModule),
         typeof(AbpBackgroundJobsHangfireModule)
@@ -302,6 +304,7 @@ namespace Lion.AbpPro
             {
                 context.AddAbpCap(capOptions =>
                 {
+                    capOptions.SetCapDbConnectionString(configuration["ConnectionStrings:Default"]);
                     capOptions.UseEntityFramework<AbpProDbContext>();
                     capOptions.UseRabbitMQ(option =>
                     {
@@ -326,7 +329,7 @@ namespace Lion.AbpPro
                     capOptions.UseInMemoryStorage();
                     capOptions.UseInMemoryMessageQueue();
                     var hostingEnvironment = context.Services.GetHostingEnvironment();
-                    bool auth = !hostingEnvironment.IsDevelopment();
+                    var auth = !hostingEnvironment.IsDevelopment();
                     capOptions.UseDashboard(options => { options.UseAuth = auth; });
                 });
             }

aspnet-core/services/host/Lion.AbpPro.HttpApi.Host/Lion.AbpPro.HttpApi.Host.csproj

@@ -44,6 +44,7 @@
     </ItemGroup>
 
     <ItemGroup>
+        <ProjectReference Include="..\..\..\frameworks\src\Lion.AbpPro.CAP.EntityFrameworkCore\Lion.AbpPro.CAP.EntityFrameworkCore.csproj" />
         <ProjectReference Include="..\..\..\frameworks\src\Lion.AbpPro.CAP\Lion.AbpPro.CAP.csproj" />
         <ProjectReference Include="..\..\..\shared\Lion.AbpPro.Shared.Hosting.Microservices\Lion.AbpPro.Shared.Hosting.Microservices.csproj" />
         <ProjectReference Include="..\..\src\Lion.AbpPro.Application\Lion.AbpPro.Application.csproj" />