使用Service Bus Trigger的Azure Functions批量完成消息时遇异常

我正在实现一个队列流程：API写入消息到Azure Service Bus队列，通过Azure Functions出队后批量插入数据库。为提升性能开启批量处理，但手动完成多条消息时遇到异常。

设置autoCompleteMessages: false后，单条消息处理代码可正常运行：

[FunctionName("BatchProcessingFunc")]
public async Task Run(
    [ServiceBusTrigger("eventbatches", Connection = "QUEUE_CONNECTIONSTRING")] Message myQueueItem, 
    MessageReceiver messageReceiver, 
    ILogger log)
{
    log.LogInformation($"C# ServiceBus queue trigger function processed message: {myQueueItem}");
    await messageReceiver.CompleteAsync(myQueueItem.SystemProperties.LockToken);        
}

但将参数改为Message[] myQueueItem批量接收后，程序抛出Microsoft.Azure.ServiceBus.MessageLockLostException，异常代码如下：

[FunctionName("BatchProcessingFunc")]
public async Task Run(
    [ServiceBusTrigger("eventbatches", Connection = "QUEUE_CONNECTIONSTRING")] Message[] myQueueItem, 
    MessageReceiver messageReceiver, 
    ILogger log)
{
    foreach(var message in myQueueItem)
    {
        log.LogInformation($"C# ServiceBus queue trigger function processed message: {message}");
        await messageReceiver.CompleteAsync(message.SystemProperties.LockToken);
    }           
}

完整异常信息：

Microsoft.Azure.ServiceBus.MessageLockLostException: The lock supplied is invalid. Either the lock expired, or the message has already been removed from the queue, or was received by a different receiver instance.
   at Microsoft.Azure.ServiceBus.Core.MessageReceiver.DisposeMessagesAsync(IEnumerable`1 lockTokens, Outcome outcome)
   at Microsoft.Azure.ServiceBus.RetryPolicy.RunOperation(Func`1 operation, TimeSpan operationTimeout)
   at Microsoft.Azure.ServiceBus.RetryPolicy.RunOperation(Func`1 operation, TimeSpan operationTimeout)
   at Microsoft.Azure.ServiceBus.Core.MessageReceiver.CompleteAsync(IEnumerable`1 lockTokens)
   at Microsoft.Azure.WebJobs.ServiceBus.Listeners.ServiceBusListener.<>c__DisplayClass35_0.<<StartMessageBatchReceiver>b__0>d.MoveNext()

观察到CompleteAsync首次执行成功，但函数会再次尝试完成消息，导致因消息已被处理而失败。由于批量写入中可能部分成功、部分失败，需要自行控制消息的完成或放弃操作。

当使用批量接收模式（Message[]）时，Azure Functions ServiceBus触发器的底层实现会在函数执行完成后，自动尝试批量完成所有接收到的消息——即使你已经手动调用了CompleteAsync。这就导致了重复完成操作，触发MessageLockLostException（因为消息锁已经被释放，消息已被移除）。

要完全自定义消息的完成/放弃逻辑，需要禁用触发器的自动批量完成行为，同时调整批量接收的配置：

1. 配置触发器的批量接收和自动完成 #

在host.json中明确配置ServiceBus触发器的批量参数，并确保autoCompleteMessages为false：

{
  "version": "2.0",
  "extensions": {
    "serviceBus": {
      "batchOptions": {
        "maxMessageCount": 100, // 单次批量接收的最大消息数
        "operationTimeout": "00:01:00"
      },
      "autoCompleteMessages": false
    }
  }
}

2. 批量完成消息（优化性能） #

避免循环调用单条CompleteAsync，改用批量完成接口，同时确保函数执行后触发器不会重复操作：

[FunctionName("BatchProcessingFunc")]
public async Task Run(
    [ServiceBusTrigger("eventbatches", Connection = "QUEUE_CONNECTIONSTRING")] Message[] myQueueItem, 
    MessageReceiver messageReceiver, 
    ILogger log)
{
    var successfulLockTokens = new List<string>();
    var failedMessages = new List<Message>();

    foreach(var message in myQueueItem)
    {
        log.LogInformation($"Processing message: {message.MessageId}");
        try
        {
            // 模拟数据库插入操作
            await InsertIntoDatabaseAsync(message);
            successfulLockTokens.Add(message.SystemProperties.LockToken);
        }
        catch(Exception ex)
        {
            log.LogError(ex, $"Failed to process message {message.MessageId}");
            failedMessages.Add(message);
        }
    }

    // 批量完成成功处理的消息
    if(successfulLockTokens.Any())
    {
        await messageReceiver.CompleteAsync(successfulLockTokens);
    }

    // 对失败消息执行放弃/死信操作（根据业务逻辑选择）
    foreach(var failedMsg in failedMessages)
    {
        // 放弃消息，让其重新进入队列
        // await messageReceiver.AbandonAsync(failedMsg.SystemProperties.LockToken);
        
        // 或者将消息移入死信队列
        await messageReceiver.DeadLetterAsync(failedMsg.SystemProperties.LockToken, "ProcessingFailed", "Failed to insert into database");
    }
}

3. 关键注意事项 #

• 锁超时时间：确保批量处理的总耗时不超过消息的锁超时时间（默认30秒），如果处理耗时较长，需在队列配置或触发器中延长锁超时，或者在处理过程中调用RenewLockAsync刷新锁。
• 幂等性：由于失败消息可能重新入队，数据库插入逻辑需要保证幂等性，避免重复插入。
• 批量操作优势：使用CompleteAsync(IEnumerable<string>)批量完成比单条调用性能更高，减少ServiceBus的API调用次数。

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