From fa9777bd087df37ff45ac8b6ad60b4401f75a435 Mon Sep 17 00:00:00 2001 From: zzh <1368735060@qq.com> Date: Sat, 10 Oct 2026 19:01:15 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20SqlSugar=20=E4=B8=8E=20EF=20Outbox=20?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E5=86=B2=E7=AA=81=E6=97=B6=E5=9B=9E=E8=90=BD?= =?UTF-8?q?=E7=9F=AD=E4=BA=8B=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 发布集成事件前先挂 Outbox 上下文；连接不匹配时 RequiresNew 写 Outbox，避免预报等按钮报 transaction not associated。 Co-authored-by: Cursor --- .../Messaging/OmsIntegrationEventPublisher.cs | 73 +++++++++++++++---- .../Orders/Batch/OrderBatchJobCoordinator.cs | 8 ++ .../Services/Orders/OrderService.BatchJob.cs | 29 ++++---- ...OrderBatchJobCoordinatorGovernanceTests.cs | 4 + 4 files changed, 86 insertions(+), 28 deletions(-) diff --git a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs index 0f12f49..956bc1e 100644 --- a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs +++ b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs @@ -1,3 +1,4 @@ +using System.Data; using Fake.EntityFrameworkCore; using Fake.EventBus.Distributed; using Fake.EventBus.IntegrationEventLog; @@ -9,6 +10,7 @@ namespace Oms.Application.Messaging; /// /// 可靠发布 IntegrationEvent：事务 UoW 内写入 Outbox（由 Fake.EventBus.IntegrationEventLog 投递）。 +/// SqlSugar 与 EF 默认各持连接；若环境事务已由 SqlSugar 占用，则改用 RequiresNew 短事务只写 Outbox。 /// internal static class OmsIntegrationEventPublisher { @@ -21,24 +23,41 @@ internal static class OmsIntegrationEventPublisher { if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager)) { - await eventBus.PublishAsync(@event, cancellationToken); - logger.LogDebug( - "Outbox enqueued {EventName} Id={EventId} (ambient TX)", + try + { + await EnsureOutboxDbContextAsync(unitOfWorkManager); + await eventBus.PublishAsync(@event, cancellationToken); + logger.LogDebug( + "Outbox enqueued {EventName} Id={EventId} (ambient TX)", + @event.GetType().Name, + @event.Id); + return; + } + catch (Exception ex) when (IsTxConnectionMismatch(ex)) + { + // SqlSugar 已在另一条连接上开事务时，EF 无法征用该 DbTransaction。 + logger.LogWarning( + ex, + "Ambient UoW TX incompatible with EF Outbox; using short outbox TX. Event={EventName} Id={EventId}", + @event.GetType().Name, + @event.Id); + } + } + else + { + logger.LogWarning( + "Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX", @event.GetType().Name, @event.Id); - return; } - // 无环境事务时开启短事务只写 Outbox，并告警：业务落库与事件非原子。 - logger.LogWarning( - "Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX", - @event.GetType().Name, - @event.Id); - - using var uow = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional()); - // 先打开 Outbox DbContext 以产生可征用的 DbTransaction - var dbProvider = uow.ServiceProvider.GetRequiredService>(); - await dbProvider.GetDbContextAsync(); + using var uow = unitOfWorkManager.Begin(new UnitOfWorkAttribute + { + IsTransactional = true, + RequiresNew = true, + IsolationLevel = IsolationLevel.ReadCommitted + }); + await EnsureOutboxDbContextAsync(unitOfWorkManager); await eventBus.PublishAsync(@event, cancellationToken); await uow.CompleteAsync(cancellationToken); } @@ -65,4 +84,30 @@ internal static class OmsIntegrationEventPublisher return false; } } + + /// + /// 业务写库前先打开 Outbox DbContext，尽量让 EF 先占住环境事务连接，供 SqlSugar 征用。 + /// + public static Task EnsureOutboxDbContextAsync(IUnitOfWorkManager unitOfWorkManager) + { + var current = unitOfWorkManager.Current + ?? throw new InvalidOperationException("No ambient unit of work"); + var dbProvider = current.ServiceProvider + .GetRequiredService>(); + return dbProvider.GetDbContextAsync(); + } + + private static bool IsTxConnectionMismatch(Exception ex) + { + for (var e = ex; e != null; e = e.InnerException!) + { + if (e.Message.Contains("not associated with the current connection", StringComparison.OrdinalIgnoreCase) + || e.Message.Contains("associated with the current connection", StringComparison.OrdinalIgnoreCase)) + { + return true; + } + } + + return false; + } } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs index b73fa91..3cac21b 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs @@ -1,5 +1,6 @@ using Fake.Application; using Fake.DependencyInjection; +using Fake.UnitOfWork; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Newtonsoft.Json; @@ -30,6 +31,7 @@ public sealed class OrderBatchJobCoordinator( IObjectStorage fileStorage, IOrderBatchJobPublisher publisher, OrderBatchJobProcessor batchJobProcessor, + IUnitOfWorkManager unitOfWorkManager, IOptions batchOptions, ILogger logger) : IScopedDependency { @@ -76,6 +78,12 @@ public sealed class OrderBatchJobCoordinator( } } + // 先打开 EF Outbox 上下文，尽量让环境事务由 EF 占连接，避免 SqlSugar 先开事务后 Outbox 征用失败。 + if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager)) + { + await OmsIntegrationEventPublisher.EnsureOutboxDbContextAsync(unitOfWorkManager); + } + var now = DateTime.Now; var job = new OrderBatchJob { diff --git a/src/Oms.Application/Services/Orders/OrderService.BatchJob.cs b/src/Oms.Application/Services/Orders/OrderService.BatchJob.cs index 13dafbf..a467162 100644 --- a/src/Oms.Application/Services/Orders/OrderService.BatchJob.cs +++ b/src/Oms.Application/Services/Orders/OrderService.BatchJob.cs @@ -1,3 +1,4 @@ +using System.Data; using Fake.Application; using Fake.Security.Claims; using Fake.UnitOfWork; @@ -22,7 +23,7 @@ public partial class OrderService /// 创建订单批量异步任务并入队 RabbitMQ。 /// [HttpPost("/order/batch/jobs")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task CreateBatchJobAsync([FromBody] OrderBatchJobCreateInput input) { return BatchJobCoordinator.CreateAndEnqueueAsync(input, CurrentUser.GetId()); @@ -75,7 +76,7 @@ public partial class OrderService /// 异步批量创建轨迹订单（替代前端逐条调用）。 /// [HttpPost("/order/batch/track-upsert")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchTrackUpsertAsync([FromBody] OrderBatchJobCreateInput input) { input.Action = OrderBatchActionCodes.TrackUpsertSmallBag; @@ -97,7 +98,7 @@ public partial class OrderService /// 异步批量预报下游订单。 /// [HttpPost("/order/batch/downstream-forward")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchDownstreamForwardAsync([FromBody] OrderBatchJobCreateInput input) { input.Action = OrderBatchActionCodes.DownstreamForwardCreateOne; @@ -115,7 +116,7 @@ public partial class OrderService string.IsNullOrWhiteSpace(authorization) ? null : authorization); } - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] private Task EnqueueOrderBatchAsync(string action, OrderBatchJobCreateInput input) { input.Action = action; @@ -126,7 +127,7 @@ public partial class OrderService /// 异步批量支付订单。 /// [HttpPost("/order/batch/pay")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchPayAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.Pay, input); @@ -134,7 +135,7 @@ public partial class OrderService /// 异步批量预报订单。 /// [HttpPost("/order/batch/pre-declare")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchPreDeclareAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.PreDeclare, input); @@ -142,7 +143,7 @@ public partial class OrderService /// 异步批量拦截订单。 /// [HttpPost("/order/batch/intercept")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchInterceptAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.Intercept, input); @@ -150,7 +151,7 @@ public partial class OrderService /// 异步批量取消拦截。 /// [HttpPost("/order/batch/cancel-intercept")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchCancelInterceptAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.CancelIntercept, input); @@ -158,7 +159,7 @@ public partial class OrderService /// 异步批量删除订单。 /// [HttpPost("/order/batch/delete")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchDeleteAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.Delete, input); @@ -166,7 +167,7 @@ public partial class OrderService /// 异步批量打印面单。 /// [HttpPost("/order/batch/print")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchPrintAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.Print, input); @@ -174,7 +175,7 @@ public partial class OrderService /// 异步批量标记问题件。 /// [HttpPost("/order/batch/mark-problematic")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchMarkProblematicAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.MarkProblematic, input); @@ -182,7 +183,7 @@ public partial class OrderService /// 异步批量重新排货。 /// [HttpPost("/order/batch/reallocate")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchReallocateAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.Reallocate, input); @@ -190,7 +191,7 @@ public partial class OrderService /// 异步批量标记已认领。 /// [HttpPost("/order/batch/mark-claimed")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchMarkClaimedAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.MarkClaimed, input); @@ -198,7 +199,7 @@ public partial class OrderService /// 异步批量更新处理状态。 /// [HttpPost("/order/batch/handle-status-update")] - [UnitOfWork(IsTransactional = true)] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public Task BatchHandleStatusUpdateAsync([FromBody] OrderBatchJobCreateInput input) => EnqueueOrderBatchAsync(OrderBatchActionCodes.HandleStatusUpdate, input); diff --git a/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs b/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs index 37ecb2f..c48e471 100644 --- a/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs +++ b/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs @@ -1,6 +1,7 @@ using FluentAssertions; using Fake.Application; using Fake.ObjectStorage; +using Fake.UnitOfWork; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Moq; @@ -36,6 +37,8 @@ public class OrderBatchJobCoordinatorGovernanceTests actionExecutor.Object, batchOptions, NullLogger.Instance); + var unitOfWorkManager = new Mock(MockBehavior.Loose); + unitOfWorkManager.SetupGet(x => x.Current).Returns((IUnitOfWork?)null); _coordinator = new OrderBatchJobCoordinator( _batchJobRepository.Object, _orderRepository.Object, @@ -44,6 +47,7 @@ public class OrderBatchJobCoordinatorGovernanceTests _fileStorage.Object, _publisher.Object, processor, + unitOfWorkManager.Object, batchOptions, NullLogger.Instance); }