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

fix: 批量队列消费者串行处理，避免 channel 并发导致消息长期 Unacked

Browse Source 
AsyncEventingBasicConsumer + Prefetch>1 时 IModel 非线程安全，并发 Ack 易卡死；改为同队列串行消费并加固 Ack/Nack。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zzh 2026-07-18 18:22:26 +08:00
parent 2d229bbedd
commit 7bc7cdf417
3 changed files with 76 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -17,6 +17,4 @@ public sealed class OrderBatchCreateConsumer(
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueCreate;
protected override bool UseRetryLimit => true;
}

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

@ -17,6 +17,4 @@ public sealed class OrderBatchIntegrationConsumer(
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueIntegration;
protected override bool UseRetryLimit => true;
}

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

@ -16,6 +16,11 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService
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;
@ -33,11 +38,6 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService
protected abstract string QueueName { get; }
/// <summary>
/// 为 true 时，Reject+Requeue 受 <see cref="OrderBatchRabbitMqOptions.MaxRetryCount"/> 限制。
/// </summary>
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<OrderBatchJobProcessor>();
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();
}