diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchDlxMonitorHostedService.cs b/src/Oms.Application/Messaging/OrderBatchDlxMonitorHostedService.cs similarity index 93% rename from src/Oms.Application/Workers/OrderBatch/OrderBatchDlxMonitorHostedService.cs rename to src/Oms.Application/Messaging/OrderBatchDlxMonitorHostedService.cs index 657cbd7..3f7ddeb 100644 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchDlxMonitorHostedService.cs +++ b/src/Oms.Application/Messaging/OrderBatchDlxMonitorHostedService.cs @@ -9,6 +9,8 @@ namespace Oms.Application.Messaging; /// public sealed class OrderBatchDlxMonitorHostedService : BackgroundService { + private static readonly TimeSpan MonitorInterval = TimeSpan.FromMinutes(5); + private readonly IOrderBatchQueueInspector _inspector; private readonly OrderBatchRabbitMqOptions _options; private readonly ILogger _logger; @@ -30,8 +32,7 @@ public sealed class OrderBatchDlxMonitorHostedService : BackgroundService return; } - var interval = TimeSpan.FromSeconds(Math.Max(60, _options.MonitorIntervalSeconds)); - using var timer = new PeriodicTimer(interval); + using var timer = new PeriodicTimer(MonitorInterval); while (!stoppingToken.IsCancellationRequested) { try diff --git a/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs b/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs new file mode 100644 index 0000000..9228509 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs @@ -0,0 +1,43 @@ +using Fake.DependencyInjection; +using Fake.EventBus; +using Fake.UnitOfWork; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Oms.Application.Services.Orders.Batch; + +namespace Oms.Application.Messaging; + +/// +/// 订单批量集成事件处理器。消费通道/Ack/Nack/DLX/重试由 Fake.EventBus.RabbitMQ 负责。 +/// +public sealed class OrderBatchJobEventHandler( + OrderBatchJobProcessor processor, + IUnitOfWorkManager unitOfWorkManager, + IOptions options, + ILogger logger) + : IEventHandler, ITransientDependency +{ + public async Task HandleAsync(OrderBatchJobIntegrationEvent @event, CancellationToken cancellationToken) + { + if (!options.Value.Enabled) + { + logger.LogDebug( + "Order batch messaging disabled, skip event {EventId} action={Action}", + @event.Id, + @event.Action); + return; + } + + using var unitOfWork = unitOfWorkManager.Begin(); + var message = @event.ToMessage(); + var result = await processor.ProcessAsync(message, cancellationToken); + await unitOfWork.CompleteAsync(cancellationToken); + + // Fake EventBus:成功返回则 Ack;抛异常则框架内重试后 Nack(requeue=false) 进 DLX。 + if (result.Kind == OrderBatchJobProcessor.ProcessResultKind.Reject && result.Requeue) + { + throw new InvalidOperationException( + $"Order batch system failure, allow Fake EventBus retry. action={message.Action} jobItemId={message.JobItemId} messageId={message.MessageId}"); + } + } +} diff --git a/src/Oms.Application/Messaging/OrderBatchJobIntegrationEvent.cs b/src/Oms.Application/Messaging/OrderBatchJobIntegrationEvent.cs new file mode 100644 index 0000000..9798f66 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderBatchJobIntegrationEvent.cs @@ -0,0 +1,52 @@ +using Fake.EventBus.Distributed; + +namespace Oms.Application.Messaging; + +/// +/// 订单批量任务集成事件(走 Fake.EventBus.RabbitMQ)。 +/// +[Serializable] +public class OrderBatchJobIntegrationEvent : IntegrationEvent +{ + public long JobId { get; set; } + + public long JobItemId { get; set; } + + public string Action { get; set; } = string.Empty; + + public int OrderId { get; set; } + + public int? CustomerId { get; set; } + + public string? PayloadJson { get; set; } + + public int OperatorUserId { get; set; } + + public string MessageId { get; set; } = string.Empty; + + public static OrderBatchJobIntegrationEvent FromMessage(OrderBatchJobMessage message) => + new() + { + JobId = message.JobId, + JobItemId = message.JobItemId, + Action = message.Action, + OrderId = message.OrderId, + CustomerId = message.CustomerId, + PayloadJson = message.PayloadJson, + OperatorUserId = message.OperatorUserId, + MessageId = message.MessageId, + }; + + public OrderBatchJobMessage ToMessage() => + new() + { + JobId = JobId, + JobItemId = JobItemId, + Action = Action, + OrderId = OrderId, + CustomerId = CustomerId, + PayloadJson = PayloadJson, + OperatorUserId = OperatorUserId, + MessageId = MessageId, + }; +} diff --git a/src/Oms.Application/Messaging/OrderBatchJobMessage.cs b/src/Oms.Application/Messaging/OrderBatchJobMessage.cs index 3a134d2..6b2522b 100644 --- a/src/Oms.Application/Messaging/OrderBatchJobMessage.cs +++ b/src/Oms.Application/Messaging/OrderBatchJobMessage.cs @@ -1,7 +1,7 @@ namespace Oms.Application.Messaging; /// -/// 订单批量任务 RabbitMQ 消息体。 +/// 订单批量任务消息体(应用内传递;发布时映射为 IntegrationEvent)。 /// public class OrderBatchJobMessage { @@ -20,6 +20,4 @@ public class OrderBatchJobMessage public int OperatorUserId { get; set; } public string MessageId { get; set; } = string.Empty; - - public int RetryCount { get; set; } } diff --git a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs index 8e7c6db..c066098 100644 --- a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs +++ b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs @@ -1,65 +1,33 @@ -using System.Text; -using Fake.RabbitMQ; +using Fake.EventBus.Distributed; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; -using Newtonsoft.Json; namespace Oms.Application.Messaging; -public sealed class OrderBatchJobPublisher : IOrderBatchJobPublisher +/// +/// 通过 Fake 分布式事件总线发布订单批量任务。 +/// +public sealed class OrderBatchJobPublisher( + IDistributedEventBus eventBus, + IOptions options, + ILogger logger) : IOrderBatchJobPublisher { - private readonly IRabbitMqChannelPool _channelPool; - private readonly OrderBatchRabbitMqOptions _options; - private readonly ILogger _logger; + public bool IsEnabled => options.Value.Enabled; - public OrderBatchJobPublisher( - IRabbitMqChannelPool channelPool, - IOptions options, - ILogger logger) + public async Task PublishAsync(OrderBatchJobMessage message, CancellationToken cancellationToken = default) { - _channelPool = channelPool; - _options = options.Value; - _logger = logger; - } - - public bool IsEnabled => _options.Enabled; - - public Task PublishAsync(OrderBatchJobMessage message, CancellationToken cancellationToken = default) - { - if (!_options.Enabled) + if (!options.Value.Enabled) { throw new InvalidOperationException("Order batch RabbitMQ is disabled."); } - var routingKey = OrderBatchQueueTopology.ResolveRoutingKey(message.Action); - var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message)); + var @event = OrderBatchJobIntegrationEvent.FromMessage(message); + await eventBus.PublishAsync(@event, cancellationToken); - using var channelAccessor = _channelPool.Acquire("oms-order-batch-publisher"); - var channel = channelAccessor.Channel; - var props = channel.CreateBasicProperties(); - props.Persistent = true; - props.ContentType = "application/json"; - props.MessageId = message.MessageId; - props.Headers = new Dictionary - { - ["action"] = message.Action, - ["jobId"] = message.JobId, - ["jobItemId"] = message.JobItemId, - }; - - channel.BasicPublish( - exchange: OrderBatchQueueTopology.Exchange, - routingKey: routingKey, - mandatory: false, - basicProperties: props, - body: body); - - _logger.LogDebug( - "Published order batch message {MessageId} action={Action} routingKey={RoutingKey}", + logger.LogDebug( + "Published order batch event {EventId} messageId={MessageId} action={Action}", + @event.Id, message.MessageId, - message.Action, - routingKey); - - return Task.CompletedTask; + message.Action); } } diff --git a/src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs b/src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs deleted file mode 100644 index 387cb38..0000000 --- a/src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs +++ /dev/null @@ -1,78 +0,0 @@ -using Fake.RabbitMQ; -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using RabbitMQ.Client; - -namespace Oms.Application.Messaging; - -/// -/// 应用启动时声明订单批量 Exchange / Queue / Binding。 -/// -public sealed class OrderBatchQueueInitializer : IHostedService -{ - private readonly IRabbitMqChannelPool _channelPool; - private readonly OrderBatchRabbitMqOptions _options; - private readonly ILogger _logger; - - public OrderBatchQueueInitializer( - IRabbitMqChannelPool channelPool, - IOptions options, - ILogger logger) - { - _channelPool = channelPool; - _options = options.Value; - _logger = logger; - } - - public Task StartAsync(CancellationToken cancellationToken) - { - if (!_options.Enabled) - { - _logger.LogInformation("Order batch RabbitMQ disabled, skip queue initialization."); - return Task.CompletedTask; - } - - try - { - using var channelAccessor = _channelPool.Acquire("oms-order-batch-init"); - var channel = channelAccessor.Channel; - - channel.ExchangeDeclare(OrderBatchQueueTopology.DeadLetterExchange, ExchangeType.Topic, durable: true); - channel.QueueDeclare( - OrderBatchQueueTopology.QueueDeadLetter, - durable: true, - exclusive: false, - autoDelete: false); - channel.QueueBind( - OrderBatchQueueTopology.QueueDeadLetter, - OrderBatchQueueTopology.DeadLetterExchange, - OrderBatchQueueTopology.RoutingDeadLetter); - - channel.ExchangeDeclare(OrderBatchQueueTopology.Exchange, ExchangeType.Topic, durable: true); - - var queueArgs = new Dictionary - { - ["x-dead-letter-exchange"] = OrderBatchQueueTopology.DeadLetterExchange, - ["x-dead-letter-routing-key"] = OrderBatchQueueTopology.RoutingDeadLetter, - }; - - foreach (var (queue, routingKey) in OrderBatchQueueTopology.Bindings) - { - channel.QueueDeclare(queue, durable: true, exclusive: false, autoDelete: false, arguments: queueArgs); - channel.QueueBind(queue, OrderBatchQueueTopology.Exchange, routingKey); - } - - _logger.LogInformation("Order batch RabbitMQ topology initialized."); - } - catch (Exception ex) - { - _logger.LogError(ex, "Failed to initialize order batch RabbitMQ topology."); - throw; - } - - return Task.CompletedTask; - } - - public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; -} diff --git a/src/Oms.Application/Messaging/OrderBatchQueueInspector.cs b/src/Oms.Application/Messaging/OrderBatchQueueInspector.cs index ac5dbb0..2e084ef 100644 --- a/src/Oms.Application/Messaging/OrderBatchQueueInspector.cs +++ b/src/Oms.Application/Messaging/OrderBatchQueueInspector.cs @@ -1,3 +1,5 @@ +using Fake; +using Fake.EventBus.RabbitMQ; using Fake.RabbitMQ; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -13,56 +15,58 @@ public interface IOrderBatchQueueInspector Task GetStatsAsync(CancellationToken cancellationToken = default); } -public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector +/// +/// 巡检 Fake.EventBus.RabbitMQ 的工作队列与死信队列。 +/// +public sealed class OrderBatchQueueInspector( + IRabbitMqConnectionPool connectionPool, + IOptions options, + IOptions eventBusOptions, + IApplicationInfo applicationInfo, + ILogger logger) : IOrderBatchQueueInspector { - private readonly IRabbitMqChannelPool _channelPool; - private readonly OrderBatchRabbitMqOptions _options; - private readonly ILogger _logger; + /// + /// DLX 堆积告警默认阈值。 + /// + public const uint DefaultDlxAlertThreshold = 100; - public OrderBatchQueueInspector( - IRabbitMqChannelPool channelPool, - IOptions options, - ILogger logger) - { - _channelPool = channelPool; - _options = options.Value; - _logger = logger; - } - - public bool IsEnabled => _options.Enabled; + public bool IsEnabled => options.Value.Enabled; public Task GetStatsAsync(CancellationToken cancellationToken = default) { - if (!_options.Enabled) + if (!options.Value.Enabled) { return Task.FromResult(new OrderBatchDlxStatsOutput { Enabled = false, - DlxAlertThreshold = _options.DlxAlertThreshold, + DlxAlertThreshold = DefaultDlxAlertThreshold, }); } - using var channelAccessor = _channelPool.Acquire("oms-order-batch-inspector"); - var channel = channelAccessor.Channel; + var busOptions = eventBusOptions.Value; + // 与 RabbitMqEventBus 一致:QueueName 为空时用 ApplicationName + var queueName = (busOptions.QueueName ?? applicationInfo.ApplicationName).TrimEnd('.') + ".Queue"; + var dlxQueueName = "DLX." + queueName; + + // Connection 由 Fake 连接池托管,禁止 Dispose + var connection = connectionPool.Get(busOptions.ConnectionName); + using var channel = connection.CreateModel(); + var queues = new List { - InspectQueue(channel, OrderBatchQueueTopology.QueueDeadLetter), - InspectQueue(channel, OrderBatchQueueTopology.QueueIntegration), - InspectQueue(channel, OrderBatchQueueTopology.QueueCreate), - InspectQueue(channel, OrderBatchQueueTopology.QueueAccount), - InspectQueue(channel, OrderBatchQueueTopology.QueueState), - InspectQueue(channel, OrderBatchQueueTopology.QueueException), - InspectQueue(channel, OrderBatchQueueTopology.QueueDocument), + InspectQueue(channel, queueName), + InspectQueue(channel, dlxQueueName), }; - var dlxCount = queues.First(x => x.QueueName == OrderBatchQueueTopology.QueueDeadLetter).MessageCount; - _logger.LogDebug("Order batch queue stats inspected, dlxMessages={DeadLetterCount}", dlxCount); + var dlxCount = queues.First(x => x.QueueName == dlxQueueName).MessageCount; + logger.LogDebug("Order batch Fake EventBus queue stats, dlxMessages={DeadLetterCount}", dlxCount); + return Task.FromResult(new OrderBatchDlxStatsOutput { Enabled = true, DeadLetterMessageCount = dlxCount, - DlxAlertThreshold = _options.DlxAlertThreshold, - Alert = dlxCount >= _options.DlxAlertThreshold, + DlxAlertThreshold = DefaultDlxAlertThreshold, + Alert = dlxCount >= DefaultDlxAlertThreshold, Queues = queues, }); } diff --git a/src/Oms.Application/Messaging/OrderBatchQueueTopology.cs b/src/Oms.Application/Messaging/OrderBatchQueueTopology.cs deleted file mode 100644 index e61399d..0000000 --- a/src/Oms.Application/Messaging/OrderBatchQueueTopology.cs +++ /dev/null @@ -1,93 +0,0 @@ -namespace Oms.Application.Messaging; - -/// -/// RabbitMQ 交换机、队列与路由键定义。 -/// -public static class OrderBatchQueueTopology -{ - public const string Exchange = "oms.order.batch"; - public const string DeadLetterExchange = "oms.order.batch.dlx"; - - public const string QueueIntegration = "oms.order.batch.integration"; - public const string QueueCreate = "oms.order.batch.create"; - public const string QueueAccount = "oms.order.batch.account"; - public const string QueueState = "oms.order.batch.state"; - public const string QueueException = "oms.order.batch.exception"; - public const string QueueDocument = "oms.order.batch.document"; - public const string QueueDeadLetter = "oms.order.batch.dlx"; - - public const string RoutingIntegrationTrack = "integration.track-upsert"; - public const string RoutingIntegrationDownstream = "integration.downstream-forward"; - public const string RoutingCreateParcel = "create.parcel-create"; - public const string RoutingCreateImport = "create.import-row"; - public const string RoutingAccountPay = "account.pay"; - public const string RoutingStatePrefix = "state."; - public const string RoutingExceptionPrefix = "exception."; - public const string RoutingDocumentPrefix = "document."; - public const string RoutingDeadLetter = "dead"; - - public static string ResolveRoutingKey(string action) => action switch - { - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.TrackUpsertSmallBag => RoutingIntegrationTrack, - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.DownstreamForwardCreateOne => RoutingIntegrationDownstream, - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.Pay => RoutingAccountPay, - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.ParcelCreate => RoutingCreateParcel, - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.ImportRow => RoutingCreateImport, - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.PreDeclare => RoutingStatePrefix + "pre-declare", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.Intercept => RoutingStatePrefix + "intercept", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.CancelIntercept => RoutingStatePrefix + "cancel-intercept", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.Delete => RoutingStatePrefix + "delete", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.HandleStatusUpdate => RoutingStatePrefix + "handle-status-update", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.MarkProblematic => RoutingExceptionPrefix + "mark-problematic", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.MarkClaimed => RoutingExceptionPrefix + "mark-claimed", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.Print => RoutingDocumentPrefix + "print", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.GenerateLabelPdf => RoutingDocumentPrefix + "generate-label-pdf", - Domain.OrdersAggregate.Consts.OrderBatchActionCodes.Export => RoutingDocumentPrefix + "export", - _ => "unknown." + action - }; - - public static string ResolveQueueName(string routingKey) - { - if (routingKey.StartsWith("integration.", StringComparison.Ordinal)) - { - return QueueIntegration; - } - - if (routingKey.StartsWith("create.", StringComparison.Ordinal)) - { - return QueueCreate; - } - - if (routingKey.StartsWith("account.", StringComparison.Ordinal)) - { - return QueueAccount; - } - - if (routingKey.StartsWith("state.", StringComparison.Ordinal)) - { - return QueueState; - } - - if (routingKey.StartsWith("exception.", StringComparison.Ordinal)) - { - return QueueException; - } - - if (routingKey.StartsWith("document.", StringComparison.Ordinal)) - { - return QueueDocument; - } - - return QueueIntegration; - } - - public static IReadOnlyList<(string Queue, string RoutingKey)> Bindings { get; } = - [ - (QueueIntegration, "integration.*"), - (QueueCreate, "create.*"), - (QueueAccount, "account.*"), - (QueueState, "state.*"), - (QueueException, "exception.*"), - (QueueDocument, "document.*"), - ]; -} diff --git a/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs b/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs index e2b821f..baaa7bb 100644 --- a/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs +++ b/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs @@ -1,25 +1,19 @@ namespace Oms.Application.Messaging; /// -/// 订单批量队列业务配置。连接信息使用 Fake 标准 RabbitMQ 配置节(FakeRabbitMqOptions)。 +/// 订单批量消息业务开关。连接/消费/QoS/DLX 由 Fake.EventBus.RabbitMQ(RabbitMqEventBus)配置。 /// public class OrderBatchRabbitMqOptions { public const string SectionName = "OrderBatch:RabbitMq"; + /// + /// 是否启用订单批量消息发布与处理。 + /// public bool Enabled { get; set; } - public ushort PrefetchCount { get; set; } = 5; - + /// + /// 系统失败时任务项最大重试次数(业务侧),与 Fake EventBus 进程内重试相互独立。 + /// public int MaxRetryCount { get; set; } = 3; - - /// - /// DLX 队列堆积告警阈值。 - /// - public uint DlxAlertThreshold { get; set; } = 100; - - /// - /// 队列监控轮询间隔(秒)。 - /// - public int MonitorIntervalSeconds { get; set; } = 300; } diff --git a/src/Oms.Application/Oms.Application.csproj b/src/Oms.Application/Oms.Application.csproj index 015e68f..352ddee 100644 --- a/src/Oms.Application/Oms.Application.csproj +++ b/src/Oms.Application/Oms.Application.csproj @@ -18,6 +18,7 @@ + diff --git a/src/Oms.Application/OmsApplicationModule.cs b/src/Oms.Application/OmsApplicationModule.cs index 2b7341a..b1a8c56 100644 --- a/src/Oms.Application/OmsApplicationModule.cs +++ b/src/Oms.Application/OmsApplicationModule.cs @@ -1,8 +1,8 @@ using System.Diagnostics; using Fake; +using Fake.EventBus.RabbitMQ; using Fake.Modularity; using Fake.ObjectMapping.AutoMapper; -using Fake.RabbitMQ; using FreeRedis; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; @@ -24,7 +24,7 @@ namespace Oms.Application; [DependsOn( typeof(OmsDomainModule), typeof(FakeObjectMappingAutoMapperModule), - typeof(FakeRabbitMqModule) + typeof(FakeEventBusRabbitMqModule) )] public class OmsApplicationModule : FakeModule { @@ -37,14 +37,9 @@ public class OmsApplicationModule : FakeModule { options.ScanProfiles(); }); - context.Services.PostConfigure(options => - { - foreach (var factory in options.Connections.Values) - { - factory.DispatchConsumersAsync = true; - factory.AutomaticRecoveryEnabled = true; - } - }); + + // Exchange 默认 Fake.Exchange.EventBus;QueueName / Retry 等用专用节覆盖 + context.Services.Configure(configuration.GetSection("RabbitMQ:EventBus")); var redisOptions = configuration.GetSection("Redis").Get(); if (redisOptions == null || string.IsNullOrWhiteSpace(redisOptions.Host)) diff --git a/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs b/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs index 210aed9..24960a7 100644 --- a/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs +++ b/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs @@ -1,7 +1,6 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Configuration; using Oms.Application.Messaging; -using Oms.Application.Workers.OrderBatch; namespace Oms.Application; @@ -12,14 +11,8 @@ public static class OrderBatchMessagingServiceExtensions services.Configure(configuration.GetSection(OrderBatchRabbitMqOptions.SectionName)); services.AddSingleton(); services.AddSingleton(); - services.AddHostedService(); + // 消费端由 Fake.EventBus.RabbitMQ(RabbitMqEventBus HostedService)托管 services.AddHostedService(); - services.AddHostedService(); - services.AddHostedService(); - services.AddHostedService(); - services.AddHostedService(); - services.AddHostedService(); - services.AddHostedService(); return services; } } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs index b203e7c..e989bc9 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs @@ -7,6 +7,7 @@ using Oms.Application.Integrations.SmallBagTrack; using Oms.Application.Messaging; using Oms.Application.Shared; using Oms.Domain.OrdersAggregate.Consts; +using Oms.Domain.OrdersAggregate.Entities; using Oms.Domain.OrdersAggregate.Repositories; using Oms.Domain.SystemAggregate.Repositories; @@ -89,30 +90,47 @@ public sealed class OrderBatchActionExecutor( OrderBatchJobMessage message, CancellationToken cancellationToken) { - var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == message.OrderId, cancellationToken); - if (order == null) + OrderParcelCreateInput body; + Order order; + + if (!string.IsNullOrWhiteSpace(message.PayloadJson)) { - return OrderBatchActionExecuteResult.BusinessFail("订单不存在"); + body = JsonConvert.DeserializeObject(message.PayloadJson!) + ?? throw new BusinessException("下游预报消息体无效"); + + // 凭证解析仍需订单上的渠道/下游绑定字段;有 payload 时只查主表一次 + var existing = await orderRepository.FirstOrDefaultAsync(x => x.Id == message.OrderId, cancellationToken); + if (existing == null) + { + return OrderBatchActionExecuteResult.BusinessFail("订单不存在"); + } + + order = existing; } + else + { + var bundles = await orderRepository.GetDownstreamCreateOneBundlesAsync([message.OrderId], cancellationToken); + if (!bundles.TryGetValue(message.OrderId, out var bundle)) + { + return OrderBatchActionExecuteResult.BusinessFail("订单不存在"); + } + + order = bundle.Order; + body = await OrderDownstreamCreateOneBodyBuilder.BuildAsync( + bundle.Order, + bundle.Sender, + bundle.Recipient, + bundle.Declarations, + bundle.Weights, + countryRepository, + cancellationToken); + } + if (string.IsNullOrWhiteSpace(order.ServiceNo)) { return OrderBatchActionExecuteResult.BusinessFail("订单运单号为空,无法预报下游"); } - var orderIds = new[] { order.Id }; - var senders = await orderRepository.GetSendersByOrderIdsAsync(orderIds, cancellationToken); - var recipients = await orderRepository.GetRecipientsByOrderIdsAsync(orderIds, cancellationToken); - var declarations = await orderRepository.GetDeclarationsByOrderIdsAsync(orderIds, cancellationToken); - var weights = await orderRepository.GetWeightsByOrderIdsAsync(orderIds, cancellationToken); - - var body = await OrderDownstreamCreateOneBodyBuilder.BuildAsync( - order, - senders.FirstOrDefault(), - recipients.FirstOrDefault(), - declarations, - weights, - countryRepository); - var creds = await downstreamCredentialResolver.ResolveAsync(order, cancellationToken); var forward = await itfxForwardClient.ForwardCreateOrderOneAsync( creds.SupplierCode, diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs index 219b3ba..7de45c3 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs @@ -11,6 +11,7 @@ using Oms.Domain.OrdersAggregate.Consts; using System.Linq.Expressions; using Oms.Domain.OrdersAggregate.Entities; using Oms.Domain.OrdersAggregate.Repositories; +using Oms.Domain.SystemAggregate.Repositories; namespace Oms.Application.Services.Orders.Batch; @@ -20,6 +21,7 @@ namespace Oms.Application.Services.Orders.Batch; public sealed class OrderBatchJobCoordinator( IOrderBatchJobRepository batchJobRepository, IOrderRepository orderRepository, + ICountryRepository countryRepository, IOrderBatchJobPublisher publisher, ILogger logger) : IScopedDependency { @@ -86,10 +88,26 @@ public sealed class OrderBatchJobCoordinator( job.Status = OrderBatchJobStatusCodes.Running; await batchJobRepository.UpdateJobAsync(job, cancellationToken); + var isDownstreamForward = string.Equals( + action, + OrderBatchActionCodes.DownstreamForwardCreateOne, + StringComparison.OrdinalIgnoreCase); + + // 下游预报:入队前批量装好 createOrderOne body,消费端不再按表连打 5 次 + Dictionary? downstreamPayloadByOrderId = null; + if (isDownstreamForward) + { + downstreamPayloadByOrderId = await BuildDownstreamPayloadsAsync(orderIds, cancellationToken); + } + + var ordersById = (await orderRepository.GetListAsync(x => orderIds.Contains(x.Id))) + .ToDictionary(x => x.Id); + var items = new List(orderIds.Count); foreach (var orderId in orderIds) { - var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == orderId, cancellationToken); + ordersById.TryGetValue(orderId, out var order); + var itemPayload = downstreamPayloadByOrderId?.GetValueOrDefault(orderId) ?? input.PayloadJson; var item = new OrderBatchJobItem { JobId = job.Id, @@ -98,7 +116,7 @@ public sealed class OrderBatchJobCoordinator( Action = action, Status = OrderBatchJobItemStatusCodes.Pending, MessageId = $"{job.Id}:{orderId}:{action}", - PayloadJson = input.PayloadJson, + PayloadJson = itemPayload, CreateUserId = operatorUserId, CreateTime = now, UpdateUserId = operatorUserId, @@ -111,7 +129,7 @@ public sealed class OrderBatchJobCoordinator( foreach (var item in items) { - var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == item.OrderId, cancellationToken); + ordersById.TryGetValue(item.OrderId, out var order); var message = new OrderBatchJobMessage { JobId = job.Id, @@ -119,7 +137,7 @@ public sealed class OrderBatchJobCoordinator( Action = action, OrderId = item.OrderId, CustomerId = order?.CustomerId, - PayloadJson = input.PayloadJson, + PayloadJson = item.PayloadJson, OperatorUserId = operatorUserId, MessageId = item.MessageId ?? $"{job.Id}:{item.OrderId}:{action}", }; @@ -502,4 +520,26 @@ public sealed class OrderBatchJobCoordinator( Total = total, }; } + + private async Task> BuildDownstreamPayloadsAsync( + IReadOnlyList orderIds, + CancellationToken cancellationToken) + { + var bundles = await orderRepository.GetDownstreamCreateOneBundlesAsync(orderIds, cancellationToken); + var result = new Dictionary(bundles.Count); + foreach (var (orderId, bundle) in bundles) + { + var body = await OrderDownstreamCreateOneBodyBuilder.BuildAsync( + bundle.Order, + bundle.Sender, + bundle.Recipient, + bundle.Declarations, + bundle.Weights, + countryRepository, + cancellationToken); + result[orderId] = JsonConvert.SerializeObject(body); + } + + return result; + } } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs index b9c9e7e..cbd99e7 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs @@ -1,4 +1,3 @@ -using System.Text; using Fake.Application; using Fake.DependencyInjection; using Microsoft.Extensions.Logging; @@ -57,7 +56,17 @@ public sealed class OrderBatchJobProcessor( if (message.JobItemId <= 0) { - return result.Success ? ProcessResult.Ack : ProcessResult.Reject(result.IsBusinessFailure); + // fire-and-forget(如轨迹同步):失败也不 requeue,避免空转打爆 DB + if (!result.Success) + { + logger.LogWarning( + "Fire-and-forget job failed action={Action} orderId={OrderId} reason={Reason}", + message.Action, + message.OrderId, + result.FailReason); + } + + return ProcessResult.Ack; } item = await batchJobRepository.GetItemByIdAsync((int)message.JobItemId, cancellationToken); @@ -138,12 +147,6 @@ public sealed class OrderBatchJobProcessor( return ProcessResult.Reject(isBusinessFailure: false); } - public static OrderBatchJobMessage? Deserialize(ReadOnlyMemory body) - { - var json = Encoding.UTF8.GetString(body.Span); - return JsonConvert.DeserializeObject(json); - } - private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken) { var job = await batchJobRepository.GetByIdAsync(jobId, cancellationToken); diff --git a/src/Oms.Application/Services/Orders/OrderService.DownstreamForward.cs b/src/Oms.Application/Services/Orders/OrderService.DownstreamForward.cs index ed1d862..28194bb 100644 --- a/src/Oms.Application/Services/Orders/OrderService.DownstreamForward.cs +++ b/src/Oms.Application/Services/Orders/OrderService.DownstreamForward.cs @@ -65,21 +65,20 @@ public partial class OrderService private async Task<(Domain.OrdersAggregate.Entities.Order order, OrderParcelCreateInput body)> LoadDownstreamCreateOneContextAsync( int orderId) { - var order = await GetOrderOrThrowAsync(orderId); - var orderIds = new[] { order.Id }; - var senders = await OrderRepository.GetSendersByOrderIdsAsync(orderIds); - var recipients = await OrderRepository.GetRecipientsByOrderIdsAsync(orderIds); - var declarations = await OrderRepository.GetDeclarationsByOrderIdsAsync(orderIds); - var weights = await OrderRepository.GetWeightsByOrderIdsAsync(orderIds); + var bundles = await OrderRepository.GetDownstreamCreateOneBundlesAsync([orderId]); + if (!bundles.TryGetValue(orderId, out var bundle)) + { + throw new BusinessException("订单不存在"); + } var body = await OrderDownstreamCreateOneBodyBuilder.BuildAsync( - order, - senders.FirstOrDefault(), - recipients.FirstOrDefault(), - declarations, - weights, + bundle.Order, + bundle.Sender, + bundle.Recipient, + bundle.Declarations, + bundle.Weights, CountryRepository); - return (order, body); + return (bundle.Order, body); } } diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs deleted file mode 100644 index 90008aa..0000000 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs +++ /dev/null @@ -1,20 +0,0 @@ -using Fake.RabbitMQ; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using Oms.Application.Messaging; - -namespace Oms.Application.Workers.OrderBatch; - -/// -/// 消费创建类队列(Excel 导入行、小包批量创建)。 -/// -public sealed class OrderBatchCreateConsumer( - IRabbitMqChannelPool channelPool, - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) -{ - protected override string QueueName => OrderBatchQueueTopology.QueueCreate; -} diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs deleted file mode 100644 index 8d32cac..0000000 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs +++ /dev/null @@ -1,20 +0,0 @@ -using Fake.RabbitMQ; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using Oms.Application.Messaging; - -namespace Oms.Application.Workers.OrderBatch; - -/// -/// 消费外部集成队列(轨迹同步、下游预报)。 -/// -public sealed class OrderBatchIntegrationConsumer( - IRabbitMqChannelPool channelPool, - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) -{ - protected override string QueueName => OrderBatchQueueTopology.QueueIntegration; -} diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs deleted file mode 100644 index d822b45..0000000 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs +++ /dev/null @@ -1,230 +0,0 @@ -using Fake.RabbitMQ; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using Oms.Application.Messaging; -using Oms.Application.Services.Orders.Batch; -using RabbitMQ.Client; -using RabbitMQ.Client.Events; - -namespace Oms.Application.Workers.OrderBatch; - -public abstract class OrderBatchQueueConsumerBase : BackgroundService -{ - private readonly IRabbitMqChannelPool _channelPool; - private readonly OrderBatchRabbitMqOptions _options; - private readonly IServiceScopeFactory _scopeFactory; - private readonly ILogger _logger; - /// - /// RabbitMQ.Client 的 IModel 非线程安全;AsyncEventingBasicConsumer + Prefetch>1 - /// 会并发进 Received,若同时 Ack/业务访问同一 channel,容易卡死并导致消息长期 Unacked。 - /// - private readonly SemaphoreSlim _processGate = new(1, 1); - private IChannelAccessor? _channelAccessor; - private IModel? _channel; - - protected OrderBatchQueueConsumerBase( - IRabbitMqChannelPool channelPool, - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - { - _channelPool = channelPool; - _options = options.Value; - _scopeFactory = scopeFactory; - _logger = logger; - } - - protected abstract string QueueName { get; } - - protected override Task ExecuteAsync(CancellationToken stoppingToken) - { - if (!_options.Enabled) - { - _logger.LogInformation( - "Order batch RabbitMQ disabled, consumer for {Queue} not started.", - QueueName); - return Task.CompletedTask; - } - - _channelAccessor = _channelPool.Acquire(QueueName); - _channel = _channelAccessor.Channel; - // Prefetch 仅表示「最多同时 Unacked 多少条」;实际处理仍串行,避免 channel 并发。 - var prefetch = _options.PrefetchCount == 0 ? (ushort)1 : _options.PrefetchCount; - _channel.BasicQos(0, prefetch, false); - - var consumer = new AsyncEventingBasicConsumer(_channel); - consumer.Received += OnReceivedAsync; - - _channel.BasicConsume(queue: QueueName, autoAck: false, consumer: consumer); - - _logger.LogInformation( - "Order batch consumer started on {Queue} (prefetch={Prefetch}, serial processing).", - QueueName, - prefetch); - - stoppingToken.Register(() => - { - try - { - _channelPool.Release(QueueName); - } - catch (Exception ex) - { - _logger.LogDebug(ex, "Error releasing RabbitMQ consumer {Queue}.", QueueName); - } - }); - - return Task.CompletedTask; - } - - private async Task OnReceivedAsync(object sender, BasicDeliverEventArgs ea) - { - if (_channel == null) - { - return; - } - - await _processGate.WaitAsync().ConfigureAwait(false); - try - { - await ProcessDeliveredAsync(ea).ConfigureAwait(false); - } - finally - { - _processGate.Release(); - } - } - - private async Task ProcessDeliveredAsync(BasicDeliverEventArgs ea) - { - if (_channel == null) - { - return; - } - - try - { - var message = OrderBatchJobProcessor.Deserialize(ea.Body); - if (message == null) - { - SafeAck(ea.DeliveryTag); - return; - } - - _logger.LogDebug( - "Consuming {Queue} action={Action} jobId={JobId} itemId={JobItemId} deliveryTag={DeliveryTag}", - QueueName, - message.Action, - message.JobId, - message.JobItemId, - ea.DeliveryTag); - - using var scope = _scopeFactory.CreateScope(); - var processor = scope.ServiceProvider.GetRequiredService(); - var result = await processor.ProcessAsync(message).ConfigureAwait(false); - - switch (result.Kind) - { - case OrderBatchJobProcessor.ProcessResultKind.Ack: - SafeAck(ea.DeliveryTag); - break; - case OrderBatchJobProcessor.ProcessResultKind.Reject: - if (result.Requeue) - { - // 瞬时失败重投;次数耗尽时 Processor 会返回 Ack(DB 已记 Failed) - SafeNack(ea.DeliveryTag, requeue: true); - } - else - { - // 业务失败:Ack 丢掉消息,状态已在 DB 记 Failed - SafeAck(ea.DeliveryTag); - } - - break; - } - } - catch (Exception ex) - { - _logger.LogError(ex, "Order batch consumer failed on queue {Queue}.", QueueName); - SafeNack(ea.DeliveryTag, requeue: true); - } - } - - private void SafeAck(ulong deliveryTag) - { - try - { - _channel?.BasicAck(deliveryTag, false); - } - catch (Exception ex) - { - _logger.LogError(ex, "BasicAck failed on {Queue}, deliveryTag={DeliveryTag}", QueueName, deliveryTag); - } - } - - private void SafeNack(ulong deliveryTag, bool requeue) - { - try - { - _channel?.BasicNack(deliveryTag, false, requeue); - } - catch (Exception ex) - { - _logger.LogError( - ex, - "BasicNack failed on {Queue}, deliveryTag={DeliveryTag}, requeue={Requeue}", - QueueName, - deliveryTag, - requeue); - } - } - - public override void Dispose() - { - _processGate.Dispose(); - _channelAccessor?.Dispose(); - base.Dispose(); - } -} - -public sealed class OrderBatchAccountConsumer( - IRabbitMqChannelPool channelPool, - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) -{ - protected override string QueueName => OrderBatchQueueTopology.QueueAccount; -} - -public sealed class OrderBatchStateConsumer( - IRabbitMqChannelPool channelPool, - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) -{ - protected override string QueueName => OrderBatchQueueTopology.QueueState; -} - -public sealed class OrderBatchExceptionConsumer( - IRabbitMqChannelPool channelPool, - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) -{ - protected override string QueueName => OrderBatchQueueTopology.QueueException; -} - -public sealed class OrderBatchDocumentConsumer( - IRabbitMqChannelPool channelPool, - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) -{ - protected override string QueueName => OrderBatchQueueTopology.QueueDocument; -} diff --git a/src/Oms.Domain/OrdersAggregate/OrderDownstreamCreateOneBundle.cs b/src/Oms.Domain/OrdersAggregate/OrderDownstreamCreateOneBundle.cs new file mode 100644 index 0000000..6931376 --- /dev/null +++ b/src/Oms.Domain/OrdersAggregate/OrderDownstreamCreateOneBundle.cs @@ -0,0 +1,19 @@ +using Oms.Domain.OrdersAggregate.Entities; + +namespace Oms.Domain.OrdersAggregate; + +/// +/// 下游 createOrderOne 组装所需的订单主表 + 关联子表(一次仓储加载)。 +/// +public sealed class OrderDownstreamCreateOneBundle +{ + public required Order Order { get; init; } + + public OrderSender? Sender { get; init; } + + public OrderRecipient? Recipient { get; init; } + + public IReadOnlyList Declarations { get; init; } = []; + + public IReadOnlyList Weights { get; init; } = []; +} diff --git a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs index df46cf2..f720371 100644 --- a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs +++ b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs @@ -35,6 +35,13 @@ public interface IOrderRepository : IRepository Task> GetWeightsByOrderIdsAsync(IEnumerable orderIds, CancellationToken cancellationToken = default); + /// + /// 一次加载下游预报所需订单主表与关联子表(避免消费端按表连打 5 次)。 + /// + Task> GetDownstreamCreateOneBundlesAsync( + IEnumerable orderIds, + CancellationToken cancellationToken = default); + Task InsertSenderAsync(OrderSender sender, CancellationToken cancellationToken = default); Task InsertRecipientAsync(OrderRecipient recipient, CancellationToken cancellationToken = default); diff --git a/src/Oms.SqlSugarCore/Orders/OrderRepository.cs b/src/Oms.SqlSugarCore/Orders/OrderRepository.cs index f18e682..91074e7 100644 --- a/src/Oms.SqlSugarCore/Orders/OrderRepository.cs +++ b/src/Oms.SqlSugarCore/Orders/OrderRepository.cs @@ -206,6 +206,62 @@ public class OrderRepository : OmsSqlSugarRepository, IOrderRepository, I .ToListAsync(cancellationToken); } + public async Task> GetDownstreamCreateOneBundlesAsync( + IEnumerable orderIds, + CancellationToken cancellationToken = default) + { + var ids = orderIds.Where(x => x > 0).Distinct().ToArray(); + if (ids.Length == 0) + { + return new Dictionary(); + } + + var db = await GetDbContextAsync(cancellationToken); + var orders = await db.Queryable() + .Where(x => ids.Contains(x.Id)) + .ToListAsync(cancellationToken); + if (orders.Count == 0) + { + return new Dictionary(); + } + + var foundIds = orders.Select(x => x.Id).ToArray(); + var senders = await db.Queryable() + .Where(x => foundIds.Contains(x.OrderId)) + .ToListAsync(cancellationToken); + var recipients = await db.Queryable() + .Where(x => foundIds.Contains(x.OrderId)) + .ToListAsync(cancellationToken); + var declarations = await db.Queryable() + .Where(x => foundIds.Contains(x.OrderId)) + .ToListAsync(cancellationToken); + var weights = await db.Queryable() + .Where(x => foundIds.Contains(x.OrderId)) + .ToListAsync(cancellationToken); + + var sendersByOrder = senders.GroupBy(x => x.OrderId).ToDictionary(g => g.Key, g => g.First()); + var recipientsByOrder = recipients.GroupBy(x => x.OrderId).ToDictionary(g => g.Key, g => g.First()); + var declarationsByOrder = declarations.GroupBy(x => x.OrderId) + .ToDictionary(g => g.Key, g => (IReadOnlyList)g.ToList()); + var weightsByOrder = weights.GroupBy(x => x.OrderId) + .ToDictionary(g => g.Key, g => (IReadOnlyList)g.ToList()); + + var result = new Dictionary(orders.Count); + foreach (var order in orders) + { + result[order.Id] = new OrderDownstreamCreateOneBundle + { + Order = order, + Sender = sendersByOrder.GetValueOrDefault(order.Id), + Recipient = recipientsByOrder.GetValueOrDefault(order.Id), + Declarations = declarationsByOrder.GetValueOrDefault(order.Id) ?? [], + Weights = weightsByOrder.GetValueOrDefault(order.Id) ?? [], + }; + } + + return result; + } + public async Task InsertSenderAsync(OrderSender sender, CancellationToken cancellationToken = default) { var db = await GetDbContextAsync(cancellationToken); diff --git a/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs b/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs index db4d4d3..95448cb 100644 --- a/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs +++ b/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs @@ -8,6 +8,7 @@ using Oms.Application.Services.Orders.Batch; using Oms.Domain.OrdersAggregate.Consts; using Oms.Domain.OrdersAggregate.Entities; using Oms.Domain.OrdersAggregate.Repositories; +using Oms.Domain.SystemAggregate.Repositories; namespace Oms.Tests; @@ -15,6 +16,7 @@ public class OrderBatchJobCoordinatorGovernanceTests { private readonly Mock _batchJobRepository = new(MockBehavior.Strict); private readonly Mock _orderRepository = new(MockBehavior.Loose); + private readonly Mock _countryRepository = new(MockBehavior.Loose); private readonly Mock _publisher = new(MockBehavior.Strict); private readonly OrderBatchJobCoordinator _coordinator; @@ -24,6 +26,7 @@ public class OrderBatchJobCoordinatorGovernanceTests _coordinator = new OrderBatchJobCoordinator( _batchJobRepository.Object, _orderRepository.Object, + _countryRepository.Object, _publisher.Object, NullLogger.Instance); } diff --git a/src/Oms.Tests/OrderBatchJobProcessorTests.cs b/src/Oms.Tests/OrderBatchJobProcessorTests.cs index 5ef6f05..28e11aa 100644 --- a/src/Oms.Tests/OrderBatchJobProcessorTests.cs +++ b/src/Oms.Tests/OrderBatchJobProcessorTests.cs @@ -101,6 +101,30 @@ public class OrderBatchJobProcessorTests Times.Never); } + [Fact] + public async Task ProcessAsync_FireAndForgetFailure_ShouldAckWithoutRequeue() + { + var message = new OrderBatchJobMessage + { + JobId = 0, + JobItemId = 0, + Action = OrderBatchActionCodes.TrackUpsertSmallBag, + OrderId = 99, + MessageId = "track:99:x", + }; + _actionExecutor + .Setup(x => x.ExecuteAsync(message, It.IsAny())) + .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时")); + + var result = await _processor.ProcessAsync(message); + + result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack); + result.Requeue.Should().BeFalse(); + _batchJobRepository.Verify( + x => x.GetItemByIdAsync(It.IsAny(), It.IsAny()), + Times.Never); + } + private void SetupItemLookup(OrderBatchJobItem item) { _batchJobRepository diff --git a/src/SimpleAdmin.Api/appsettings.Development.json b/src/SimpleAdmin.Api/appsettings.Development.json index 4edc5b8..5915529 100644 --- a/src/SimpleAdmin.Api/appsettings.Development.json +++ b/src/SimpleAdmin.Api/appsettings.Development.json @@ -105,12 +105,15 @@ "Password": "oms123", "VirtualHost": "/oms" } + }, + "EventBus": { + "QueueName": "nanafa.oms", + "RetryCount": 5 } }, "OrderBatch": { "RabbitMq": { "Enabled": false, - "PrefetchCount": 5, "MaxRetryCount": 3 } }