Azure Service Bus长运行消息锁失效导致完成报错求助

我们有一条最长需1小时处理的消息，要求仅被处理一次。当前遇到的问题是，在尝试完成消息时，该消息已对其他处理器可用。

当前Service Bus配置 #

var options = new ServiceBusProcessorOptions
{
    AutoCompleteMessages = false,

    MaxConcurrentCalls = 1,
    PrefetchCount = 0,
    ReceiveMode = ServiceBusReceiveMode.PeekLock,
    MaxAutoLockRenewalDuration = TimeSpan.FromHours(2),

};

当前消息处理逻辑 #

private async Task HandleMessageAsync(ProcessMessageEventArgs processMessageEventArgs)
{
    try
    {
        var rawMessageBody = Encoding.UTF8.GetString(processMessageEventArgs.Message.Body);
        _logger.LogInformation("Received message {MessageId} with body {MessageBody}",
            processMessageEventArgs.Message.MessageId, rawMessageBody);

        var repoRequest = JsonConvert.DeserializeObject<TMessage>(rawMessageBody);
        if (repoRequest != null)
        {
            await ProcessMessage(repoRequest, processMessageEventArgs.Message.MessageId,
                processMessageEventArgs.Message.ApplicationProperties,
                processMessageEventArgs.CancellationToken);
        }
        else
        {
            _logger.LogError(
                "Unable to deserialize to message contract {ContractName} for message {MessageBody}",
                typeof(TMessage), rawMessageBody);
        }

        _logger.LogInformation("Message {MessageId} processed", processMessageEventArgs.Message.MessageId);

        await processMessageEventArgs.CompleteMessageAsync(processMessageEventArgs.Message);
    }
    catch (Exception ex)
    {
        await processMessageEventArgs.AbandonMessageAsync(processMessageEventArgs.Message);
        _logger.LogError(ex, "Unable to handle message");
    }
}

异常现象 #

处理过程中持续出现ServiceBusReceiver.RenewMessageLock异常，但消息仍在处理；最终调用await processMessageEventArgs.CompleteMessageAsync(processMessageEventArgs.Message);时，触发MessageLockLost类型的ServiceBusException，提示锁无效。已知MaxAutoLockRenewalDuration已设为2小时，但Azure无法保证所有场景下的续期。

1. 延长初始消息锁时长 #

默认消息锁时长为30秒，频繁续期容易引发失败。直接将队列/主题的初始锁时长设置为略长于消息最大处理时间（比如1小时10分钟），减少续期次数，从根源降低续期失败概率。

注意：此设置需在Azure Portal或ARM模板中修改队列/主题的Lock Duration属性，代码无法覆盖该配置。

2. 实现幂等处理，确保消息仅执行一次 #

锁丢失后消息可能被其他实例接收，必须通过幂等机制避免重复处理：

• 以消息的MessageId或业务唯一ID作为标识
• 在处理前先校验该标识是否已完成处理（可存入数据库或分布式缓存）
• 处理完成后标记该标识为已处理

修改后的核心逻辑示例：

private async Task HandleMessageAsync(ProcessMessageEventArgs processMessageEventArgs)
{
    var messageId = processMessageEventArgs.Message.MessageId;
    try
    {
        var rawMessageBody = Encoding.UTF8.GetString(processMessageEventArgs.Message.Body);
        _logger.LogInformation("Received message {MessageId} with body {MessageBody}", messageId, rawMessageBody);

        // 幂等校验：检查消息是否已处理
        if (await IsMessageProcessed(messageId))
        {
            _logger.LogInformation("Message {MessageId} already processed, skip", messageId);
            try
            {
                await processMessageEventArgs.CompleteMessageAsync(processMessageEventArgs.Message);
            }
            catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessageLockLost)
            {
                _logger.LogWarning("Message {MessageId} lock lost but already processed", messageId);
            }
            return;
        }

        var repoRequest = JsonConvert.DeserializeObject<TMessage>(rawMessageBody);
        if (repoRequest != null)
        {
            await ProcessMessage(repoRequest, messageId,
                processMessageEventArgs.Message.ApplicationProperties,
                processMessageEventArgs.CancellationToken);
            // 标记消息为已处理
            await MarkMessageAsProcessed(messageId);
        }
        else
        {
            _logger.LogError("Failed to deserialize message {MessageId} to {Contract}", messageId, typeof(TMessage));
            // 反序列化失败直接死信，避免重复重试
            await processMessageEventArgs.DeadLetterMessageAsync(processMessageEventArgs.Message, "DeserializationError", "Invalid message format");
            return;
        }

        _logger.LogInformation("Message {MessageId} processed successfully", messageId);
        await processMessageEventArgs.CompleteMessageAsync(processMessageEventArgs.Message);
    }
    catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessageLockLost)
    {
        // 锁丢失：依赖幂等机制，后续实例处理时会自动跳过重复执行
        _logger.LogWarning("Message {MessageId} lock lost during processing", messageId);
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "Error processing message {MessageId}", messageId);
        // 业务异常直接死信，避免无效循环
        await processMessageEventArgs.DeadLetterMessageAsync(processMessageEventArgs.Message, "ProcessingError", ex.Message);
    }
}

// 示例：从数据库查询消息是否已处理
private async Task<bool> IsMessageProcessed(string messageId)
{
    // 替换为实际的存储查询逻辑，比如EF Core查询
    // return await _dbContext.ProcessedMessages.AnyAsync(m => m.MessageId == messageId);
    return false;
}

// 示例：标记消息为已处理
private async Task MarkMessageAsProcessed(string messageId)
{
    // 替换为实际的存储写入逻辑
    // _dbContext.ProcessedMessages.Add(new ProcessedMessage { MessageId = messageId, ProcessedTime = DateTime.UtcNow });
    // await _dbContext.SaveChangesAsync();
}

3. 优化自动续期或手动控制续期 #

如果自动续期仍不稳定，可手动控制续期逻辑：

• 在处理任务中启动后台定时续期任务
• 监控续期结果，一旦续期失败立即终止处理

示例手动续期逻辑：

private async Task ProcessMessage(TMessage request, string messageId, IDictionary<string, object> properties, CancellationToken cancellationToken)
{
    using var renewalCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
    var renewalTask = Task.Run(async () =>
    {
        // 每25秒续期一次（比30秒默认锁时长提前5秒）
        var renewalInterval = TimeSpan.FromSeconds(25);
        while (!renewalCts.Token.IsCancellationRequested)
        {
            try
            {
                await Task.Delay(renewalInterval, renewalCts.Token);
                await processMessageEventArgs.RenewMessageLockAsync(processMessageEventArgs.Message, renewalCts.Token);
                _logger.LogInformation("Renewed lock for message {MessageId}", messageId);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to renew lock for message {MessageId}", messageId);
                renewalCts.Cancel();
            }
        }
    }, cancellationToken);

    try
    {
        // 执行实际业务处理
        await _businessService.Execute(request, cancellationToken);
    }
    finally
    {
        renewalCts.Cancel();
        await renewalTask;
    }
}

4. 调整异常处理策略 #

原逻辑中所有异常都调用AbandonMessageAsync，会导致消息立即重回队列，引发重复处理。优化策略：

• MessageLockLostException：不做Abandon，依赖幂等机制处理后续重复消息
• 反序列化失败、业务逻辑错误：直接将消息死信，避免无效重试
• 临时异常（如网络波动）：可调用Abandon让消息延迟重试（需配合队列的Max Delivery Count设置）

内容的提问来源于stack exchange，提问作者user1112324