diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs index 8a52b02..90008aa 100644 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs +++ b/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs @@ -17,6 +17,4 @@ public sealed class OrderBatchCreateConsumer( : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) { protected override string QueueName => OrderBatchQueueTopology.QueueCreate; - - protected override bool UseRetryLimit => true; } diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs index ddaa095..8d32cac 100644 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs +++ b/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs @@ -17,6 +17,4 @@ public sealed class OrderBatchIntegrationConsumer( : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) { protected override string QueueName => OrderBatchQueueTopology.QueueIntegration; - - protected override bool UseRetryLimit => true; } diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs index 6845f2d..d822b45 100644 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs +++ b/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs @@ -16,6 +16,11 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService 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; @@ -33,11 +38,6 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService protected abstract string QueueName { get; } - /// - /// 为 true 时,Reject+Requeue 受 限制。 - /// - protected virtual bool UseRetryLimit => false; - protected override Task ExecuteAsync(CancellationToken stoppingToken) { if (!_options.Enabled) @@ -50,14 +50,19 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService _channelAccessor = _channelPool.Acquire(QueueName); _channel = _channelAccessor.Channel; - _channel.BasicQos(0, _options.PrefetchCount, false); + // 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}.", QueueName); + _logger.LogInformation( + "Order batch consumer started on {Queue} (prefetch={Prefetch}, serial processing).", + QueueName, + prefetch); stoppingToken.Register(() => { @@ -81,47 +86,60 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService 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) { - _channel.BasicAck(ea.DeliveryTag, false); + 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); + var result = await processor.ProcessAsync(message).ConfigureAwait(false); switch (result.Kind) { case OrderBatchJobProcessor.ProcessResultKind.Ack: - _channel.BasicAck(ea.DeliveryTag, false); + SafeAck(ea.DeliveryTag); break; case OrderBatchJobProcessor.ProcessResultKind.Reject: if (result.Requeue) { - if (UseRetryLimit) - { - if (message.RetryCount < _options.MaxRetryCount) - { - message.RetryCount++; - _channel.BasicNack(ea.DeliveryTag, false, true); - } - else - { - _channel.BasicAck(ea.DeliveryTag, false); - } - } - else - { - _channel.BasicNack(ea.DeliveryTag, false, true); - } + // 瞬时失败重投;次数耗尽时 Processor 会返回 Ack(DB 已记 Failed) + SafeNack(ea.DeliveryTag, requeue: true); } else { - _channel.BasicAck(ea.DeliveryTag, false); + // 业务失败:Ack 丢掉消息,状态已在 DB 记 Failed + SafeAck(ea.DeliveryTag); } break; @@ -130,12 +148,42 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService catch (Exception ex) { _logger.LogError(ex, "Order batch consumer failed on queue {Queue}.", QueueName); - _channel.BasicNack(ea.DeliveryTag, false, true); + 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(); }