ideship/admin
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.04 #6

Merged
zzh merged 4 commits from test into master 2026-10-10 19:35:37 +08:00
Conversation 0 Commits 4 Files Changed 21 +86 -28
4 changed files with 86 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit fa9777bd08 - Show all commits

57
src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs
View File

@ -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;
/// <summary>
/// 可靠发布 IntegrationEvent：事务 UoW 内写入 Outbox（由 Fake.EventBus.IntegrationEventLog 投递）。
/// SqlSugar 与 EF 默认各持连接；若环境事务已由 SqlSugar 占用，则改用 RequiresNew 短事务只写 Outbox。
/// </summary>
internal static class OmsIntegrationEventPublisher
{
@ -21,6 +23,9 @@ internal static class OmsIntegrationEventPublisher
{
if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager))
{
try
{
await EnsureOutboxDbContextAsync(unitOfWorkManager);
await eventBus.PublishAsync(@event, cancellationToken);
logger.LogDebug(
"Outbox enqueued {EventName} Id={EventId} (ambient TX)",
@ -28,17 +33,31 @@ internal static class OmsIntegrationEventPublisher
@event.Id);
return;
}
// 无环境事务时开启短事务只写 Outbox，并告警：业务落库与事件非原子。
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);
}
using var uow = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional());
// 先打开 Outbox DbContext 以产生可征用的 DbTransaction
var dbProvider = uow.ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
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;
}
}
/// <summary>
/// 业务写库前先打开 Outbox DbContext，尽量让 EF 先占住环境事务连接，供 SqlSugar 征用。
/// </summary>
public static Task EnsureOutboxDbContextAsync(IUnitOfWorkManager unitOfWorkManager)
{
var current = unitOfWorkManager.Current
?? throw new InvalidOperationException("No ambient unit of work");
var dbProvider = current.ServiceProvider
.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
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;
}
}

8
src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs
View File

@ -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<OrderBatchRabbitMqOptions> batchOptions,
ILogger<OrderBatchJobCoordinator> 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
{

29
src/Oms.Application/Services/Orders/OrderService.BatchJob.cs
View File

@ -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。
/// </summary>
[HttpPost("/order/batch/jobs")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> CreateBatchJobAsync([FromBody] OrderBatchJobCreateInput input)
{
return BatchJobCoordinator.CreateAndEnqueueAsync(input, CurrentUser.GetId<int>());
@ -75,7 +76,7 @@ public partial class OrderService
/// 异步批量创建轨迹订单（替代前端逐条调用）。
/// </summary>
[HttpPost("/order/batch/track-upsert")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchTrackUpsertAsync([FromBody] OrderBatchJobCreateInput input)
{
input.Action = OrderBatchActionCodes.TrackUpsertSmallBag;
@ -97,7 +98,7 @@ public partial class OrderService
/// 异步批量预报下游订单。
/// </summary>
[HttpPost("/order/batch/downstream-forward")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> 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<OrderBatchJobOutput> EnqueueOrderBatchAsync(string action, OrderBatchJobCreateInput input)
{
input.Action = action;
@ -126,7 +127,7 @@ public partial class OrderService
/// 异步批量支付订单。
/// </summary>
[HttpPost("/order/batch/pay")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchPayAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Pay, input);
@ -134,7 +135,7 @@ public partial class OrderService
/// 异步批量预报订单。
/// </summary>
[HttpPost("/order/batch/pre-declare")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchPreDeclareAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.PreDeclare, input);
@ -142,7 +143,7 @@ public partial class OrderService
/// 异步批量拦截订单。
/// </summary>
[HttpPost("/order/batch/intercept")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchInterceptAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Intercept, input);
@ -150,7 +151,7 @@ public partial class OrderService
/// 异步批量取消拦截。
/// </summary>
[HttpPost("/order/batch/cancel-intercept")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchCancelInterceptAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.CancelIntercept, input);
@ -158,7 +159,7 @@ public partial class OrderService
/// 异步批量删除订单。
/// </summary>
[HttpPost("/order/batch/delete")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchDeleteAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Delete, input);
@ -166,7 +167,7 @@ public partial class OrderService
/// 异步批量打印面单。
/// </summary>
[HttpPost("/order/batch/print")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchPrintAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Print, input);
@ -174,7 +175,7 @@ public partial class OrderService
/// 异步批量标记问题件。
/// </summary>
[HttpPost("/order/batch/mark-problematic")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchMarkProblematicAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.MarkProblematic, input);
@ -182,7 +183,7 @@ public partial class OrderService
/// 异步批量重新排货。
/// </summary>
[HttpPost("/order/batch/reallocate")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchReallocateAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Reallocate, input);
@ -190,7 +191,7 @@ public partial class OrderService
/// 异步批量标记已认领。
/// </summary>
[HttpPost("/order/batch/mark-claimed")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchMarkClaimedAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.MarkClaimed, input);
@ -198,7 +199,7 @@ public partial class OrderService
/// 异步批量更新处理状态。
/// </summary>
[HttpPost("/order/batch/handle-status-update")]
[UnitOfWork(IsTransactional = true)]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task<OrderBatchJobOutput> BatchHandleStatusUpdateAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.HandleStatusUpdate, input);

4
src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs
View File

@ -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<OrderBatchJobProcessor>.Instance);
var unitOfWorkManager = new Mock<IUnitOfWorkManager>(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<OrderBatchJobCoordinator>.Instance);
}