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

refactor: 订单批量任务改走 Fake.EventBus.RabbitMQ

Browse Source 
移除自建消费者与队列拓扑，统一由事件总线发布/消费，并优化下游转发查询与配置。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
xiaolipro 2026-07-20 17:49:33 +08:00
parent 188aed1ee6
commit a4d6400318
25 changed files with 378 additions and 598 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/Oms.Application/Workers/OrderBatch/OrderBatchDlxMonitorHostedService.cs → src/Oms.Application/Messaging/OrderBatchDlxMonitorHostedService.cs
View File

@ -9,6 +9,8 @@ namespace Oms.Application.Messaging;
/// </summary>
public sealed class OrderBatchDlxMonitorHostedService : BackgroundService
{
private static readonly TimeSpan MonitorInterval = TimeSpan.FromMinutes(5);
private readonly IOrderBatchQueueInspector _inspector;
private readonly OrderBatchRabbitMqOptions _options;
private readonly ILogger<OrderBatchDlxMonitorHostedService> _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

43
src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs Normal file
View File

@ -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;
/// <summary>
/// 订单批量集成事件处理器。消费通道/Ack/Nack/DLX/重试由 Fake.EventBus.RabbitMQ 负责。
/// </summary>
public sealed class OrderBatchJobEventHandler(
OrderBatchJobProcessor processor,
IUnitOfWorkManager unitOfWorkManager,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchJobEventHandler> logger)
: IEventHandler<OrderBatchJobIntegrationEvent>, 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}");
}
}
}

52
src/Oms.Application/Messaging/OrderBatchJobIntegrationEvent.cs Normal file
View File

@ -0,0 +1,52 @@
using Fake.EventBus.Distributed;
namespace Oms.Application.Messaging;
/// <summary>
/// 订单批量任务集成事件（走 Fake.EventBus.RabbitMQ）。
/// </summary>
[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,
};
}

4
src/Oms.Application/Messaging/OrderBatchJobMessage.cs
View File

@ -1,7 +1,7 @@
namespace Oms.Application.Messaging;
/// <summary>
/// 订单批量任务 RabbitMQ 消息体。
/// 订单批量任务消息体（应用内传递；发布时映射为 IntegrationEvent）。
/// </summary>
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; }
}

66
src/Oms.Application/Messaging/OrderBatchJobPublisher.cs
View File

@ -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
{
private readonly IRabbitMqChannelPool _channelPool;
private readonly OrderBatchRabbitMqOptions _options;
private readonly ILogger<OrderBatchJobPublisher> _logger;
public OrderBatchJobPublisher(
IRabbitMqChannelPool channelPool,
/// <summary>
/// 通过 Fake 分布式事件总线发布订单批量任务。
/// </summary>
public sealed class OrderBatchJobPublisher(
IDistributedEventBus eventBus,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchJobPublisher> logger)
{
_channelPool = channelPool;
_options = options.Value;
_logger = logger;
}
ILogger<OrderBatchJobPublisher> logger) : IOrderBatchJobPublisher
{
public bool IsEnabled => options.Value.Enabled;
public bool IsEnabled => _options.Enabled;
public Task PublishAsync(OrderBatchJobMessage message, CancellationToken cancellationToken = default)
public async 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<string, object>
{
["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);
}
}

78
src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs
View File

@ -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;
/// <summary>
/// 应用启动时声明订单批量 Exchange / Queue / Binding。
/// </summary>
public sealed class OrderBatchQueueInitializer : IHostedService
{
private readonly IRabbitMqChannelPool _channelPool;
private readonly OrderBatchRabbitMqOptions _options;
private readonly ILogger<OrderBatchQueueInitializer> _logger;
public OrderBatchQueueInitializer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchQueueInitializer> 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<string, object>
{
["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;
}

64
src/Oms.Application/Messaging/OrderBatchQueueInspector.cs
View File

@ -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<OrderBatchDlxStatsOutput> GetStatsAsync(CancellationToken cancellationToken = default);
}
public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector
{
private readonly IRabbitMqChannelPool _channelPool;
private readonly OrderBatchRabbitMqOptions _options;
private readonly ILogger<OrderBatchQueueInspector> _logger;
public OrderBatchQueueInspector(
IRabbitMqChannelPool channelPool,
/// <summary>
/// 巡检 Fake.EventBus.RabbitMQ 的工作队列与死信队列。
/// </summary>
public sealed class OrderBatchQueueInspector(
IRabbitMqConnectionPool connectionPool,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchQueueInspector> logger)
{
_channelPool = channelPool;
_options = options.Value;
_logger = logger;
}
IOptions<RabbitMqEventBusOptions> eventBusOptions,
IApplicationInfo applicationInfo,
ILogger<OrderBatchQueueInspector> logger) : IOrderBatchQueueInspector
{
/// <summary>
/// DLX 堆积告警默认阈值。
/// </summary>
public const uint DefaultDlxAlertThreshold = 100;
public bool IsEnabled => _options.Enabled;
public bool IsEnabled => options.Value.Enabled;
public Task<OrderBatchDlxStatsOutput> 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<OrderBatchQueueStatOutput>
{
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,
});
}

93
src/Oms.Application/Messaging/OrderBatchQueueTopology.cs
View File

@ -1,93 +0,0 @@
namespace Oms.Application.Messaging;
/// <summary>
/// RabbitMQ 交换机、队列与路由键定义。
/// </summary>
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.*"),
];
}

20
src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs
View File

@ -1,25 +1,19 @@
namespace Oms.Application.Messaging;
/// <summary>
/// 订单批量队列业务配置。连接信息使用 Fake 标准 <c>RabbitMQ</c> 配置节（FakeRabbitMqOptions）。
/// 订单批量消息业务开关。连接/消费/QoS/DLX 由 Fake.EventBus.RabbitMQ（RabbitMqEventBus）配置。
/// </summary>
public class OrderBatchRabbitMqOptions
{
public const string SectionName = "OrderBatch:RabbitMq";
/// <summary>
/// 是否启用订单批量消息发布与处理。
/// </summary>
public bool Enabled { get; set; }
public ushort PrefetchCount { get; set; } = 5;
/// <summary>
/// 系统失败时任务项最大重试次数（业务侧），与 Fake EventBus 进程内重试相互独立。
/// </summary>
public int MaxRetryCount { get; set; } = 3;
/// <summary>
/// DLX 队列堆积告警阈值。
/// </summary>
public uint DlxAlertThreshold { get; set; } = 100;
/// <summary>
/// 队列监控轮询间隔（秒）。
/// </summary>
public int MonitorIntervalSeconds { get; set; } = 300;
}

1
src/Oms.Application/Oms.Application.csproj
View File

@ -18,6 +18,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Fake.EventBus.RabbitMQ" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectMapping.AutoMapper" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.RabbitMQ" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Rbac.Application" Version="$(FakePackageVersion)" />

15
src/Oms.Application/OmsApplicationModule.cs
View File

@ -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<OmsApplicationModule>();
});
context.Services.PostConfigure<FakeRabbitMqOptions>(options =>
{
foreach (var factory in options.Connections.Values)
{
factory.DispatchConsumersAsync = true;
factory.AutomaticRecoveryEnabled = true;
}
});
// Exchange 默认 Fake.Exchange.EventBus；QueueName / Retry 等用专用节覆盖
context.Services.Configure<RabbitMqEventBusOptions>(configuration.GetSection("RabbitMQ:EventBus"));
var redisOptions = configuration.GetSection("Redis").Get<ConnectionStringBuilder>();
if (redisOptions == null || string.IsNullOrWhiteSpace(redisOptions.Host))

9
src/Oms.Application/OrderBatchMessagingServiceExtensions.cs
View File

@ -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<OrderBatchRabbitMqOptions>(configuration.GetSection(OrderBatchRabbitMqOptions.SectionName));
services.AddSingleton<IOrderBatchJobPublisher, OrderBatchJobPublisher>();
services.AddSingleton<IOrderBatchQueueInspector, OrderBatchQueueInspector>();
services.AddHostedService<OrderBatchQueueInitializer>();
// 消费端由 Fake.EventBus.RabbitMQ（RabbitMqEventBus HostedService）托管
services.AddHostedService<OrderBatchDlxMonitorHostedService>();
services.AddHostedService<OrderBatchIntegrationConsumer>();
services.AddHostedService<OrderBatchCreateConsumer>();
services.AddHostedService<OrderBatchAccountConsumer>();
services.AddHostedService<OrderBatchStateConsumer>();
services.AddHostedService<OrderBatchExceptionConsumer>();
services.AddHostedService<OrderBatchDocumentConsumer>();
return services;
}
}

50
src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs
View File

@ -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))
{
body = JsonConvert.DeserializeObject<OrderParcelCreateInput>(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,

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

@ -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<OrderBatchJobCoordinator> 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<int, string>? 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<OrderBatchJobItem>(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<Dictionary<int, string>> BuildDownstreamPayloadsAsync(
IReadOnlyList<int> orderIds,
CancellationToken cancellationToken)
{
var bundles = await orderRepository.GetDownstreamCreateOneBundlesAsync(orderIds, cancellationToken);
var result = new Dictionary<int, string>(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;
}
}

19
src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs
View File

@ -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<byte> body)
{
var json = Encoding.UTF8.GetString(body.Span);
return JsonConvert.DeserializeObject<OrderBatchJobMessage>(json);
}
private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken)
{
var job = await batchJobRepository.GetByIdAsync(jobId, cancellationToken);

23
src/Oms.Application/Services/Orders/OrderService.DownstreamForward.cs
View File

@ -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);
}
}

20
src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs
View File

@ -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;
/// <summary>
/// 消费创建类队列（Excel 导入行、小包批量创建）。
/// </summary>
public sealed class OrderBatchCreateConsumer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchCreateConsumer> logger)
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueCreate;
}

20
src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs
View File

@ -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;
/// <summary>
/// 消费外部集成队列（轨迹同步、下游预报）。
/// </summary>
public sealed class OrderBatchIntegrationConsumer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchIntegrationConsumer> logger)
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueIntegration;
}

230
src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs
View File

@ -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;
/// <summary>
/// RabbitMQ.Client 的 IModel 非线程安全；AsyncEventingBasicConsumer + Prefetch&gt;1
/// 会并发进 Received，若同时 Ack/业务访问同一 channel，容易卡死并导致消息长期 Unacked。
/// </summary>
private readonly SemaphoreSlim _processGate = new(1, 1);
private IChannelAccessor? _channelAccessor;
private IModel? _channel;
protected OrderBatchQueueConsumerBase(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> 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<OrderBatchJobProcessor>();
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<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchAccountConsumer> logger)
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueAccount;
}
public sealed class OrderBatchStateConsumer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchStateConsumer> logger)
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueState;
}
public sealed class OrderBatchExceptionConsumer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchExceptionConsumer> logger)
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueException;
}
public sealed class OrderBatchDocumentConsumer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchDocumentConsumer> logger)
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueDocument;
}

19
src/Oms.Domain/OrdersAggregate/OrderDownstreamCreateOneBundle.cs Normal file
View File

@ -0,0 +1,19 @@
using Oms.Domain.OrdersAggregate.Entities;
namespace Oms.Domain.OrdersAggregate;
/// <summary>
/// 下游 createOrderOne 组装所需的订单主表 + 关联子表（一次仓储加载）。
/// </summary>
public sealed class OrderDownstreamCreateOneBundle
{
public required Order Order { get; init; }
public OrderSender? Sender { get; init; }
public OrderRecipient? Recipient { get; init; }
public IReadOnlyList<OrderDeclaration> Declarations { get; init; } = [];
public IReadOnlyList<OrderWeight> Weights { get; init; } = [];
}

7
src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs
View File

@ -35,6 +35,13 @@ public interface IOrderRepository : IRepository<Order>
Task<List<OrderWeight>> GetWeightsByOrderIdsAsync(IEnumerable<int> orderIds, CancellationToken cancellationToken = default);
/// <summary>
/// 一次加载下游预报所需订单主表与关联子表（避免消费端按表连打 5 次）。
/// </summary>
Task<Dictionary<int, OrderDownstreamCreateOneBundle>> GetDownstreamCreateOneBundlesAsync(
IEnumerable<int> orderIds,
CancellationToken cancellationToken = default);
Task<OrderSender> InsertSenderAsync(OrderSender sender, CancellationToken cancellationToken = default);
Task<OrderRecipient> InsertRecipientAsync(OrderRecipient recipient, CancellationToken cancellationToken = default);

56
src/Oms.SqlSugarCore/Orders/OrderRepository.cs
View File

@ -206,6 +206,62 @@ public class OrderRepository : OmsSqlSugarRepository<Order>, IOrderRepository, I
.ToListAsync(cancellationToken);
}
public async Task<Dictionary<int, OrderDownstreamCreateOneBundle>> GetDownstreamCreateOneBundlesAsync(
IEnumerable<int> orderIds,
CancellationToken cancellationToken = default)
{
var ids = orderIds.Where(x => x > 0).Distinct().ToArray();
if (ids.Length == 0)
{
return new Dictionary<int, OrderDownstreamCreateOneBundle>();
}
var db = await GetDbContextAsync(cancellationToken);
var orders = await db.Queryable<Order>()
.Where(x => ids.Contains(x.Id))
.ToListAsync(cancellationToken);
if (orders.Count == 0)
{
return new Dictionary<int, OrderDownstreamCreateOneBundle>();
}
var foundIds = orders.Select(x => x.Id).ToArray();
var senders = await db.Queryable<OrderSender>()
.Where(x => foundIds.Contains(x.OrderId))
.ToListAsync(cancellationToken);
var recipients = await db.Queryable<OrderRecipient>()
.Where(x => foundIds.Contains(x.OrderId))
.ToListAsync(cancellationToken);
var declarations = await db.Queryable<OrderDeclaration>()
.Where(x => foundIds.Contains(x.OrderId))
.ToListAsync(cancellationToken);
var weights = await db.Queryable<OrderWeight>()
.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<OrderDeclaration>)g.ToList());
var weightsByOrder = weights.GroupBy(x => x.OrderId)
.ToDictionary(g => g.Key, g => (IReadOnlyList<OrderWeight>)g.ToList());
var result = new Dictionary<int, OrderDownstreamCreateOneBundle>(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<OrderSender> InsertSenderAsync(OrderSender sender, CancellationToken cancellationToken = default)
{
var db = await GetDbContextAsync(cancellationToken);

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

@ -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<IOrderBatchJobRepository> _batchJobRepository = new(MockBehavior.Strict);
private readonly Mock<IOrderRepository> _orderRepository = new(MockBehavior.Loose);
private readonly Mock<ICountryRepository> _countryRepository = new(MockBehavior.Loose);
private readonly Mock<IOrderBatchJobPublisher> _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<OrderBatchJobCoordinator>.Instance);
}

24
src/Oms.Tests/OrderBatchJobProcessorTests.cs
View File

@ -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<CancellationToken>()))
.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<int>(), It.IsAny<CancellationToken>()),
Times.Never);
}
private void SetupItemLookup(OrderBatchJobItem item)
{
_batchJobRepository

5
src/SimpleAdmin.Api/appsettings.Development.json
View File

@ -105,12 +105,15 @@
"Password": "oms123",
"VirtualHost": "/oms"
}
},
"EventBus": {
"QueueName": "nanafa.oms",
"RetryCount": 5
}
},
"OrderBatch": {
"RabbitMq": {
"Enabled": false,
"PrefetchCount": 5,
"MaxRetryCount": 3
}
}