当一条消息无法在 ServiceBusTrigger Azure 函数中处理时，如何将同一条消息延迟处理 x 分钟？

Question

我有一个从 ServiceBus 主题读取并调用第 3 方服务的天蓝色函数。如果服务关闭，我想等待 5 分钟，然后再尝试使用相同的消息再次调用它。如何添加延迟以使 azure 函数不会放弃消息并立即将其重新拾取？

public static void Run([ServiceBusTrigger("someTopic", 
     "someSubscription", AccessRights.Manage, Connection = 
     "ServiceBusConnection")] BrokeredMessage message) 
{
     CallService(bodyOfBrokeredMessage); //service is down

     //How do I add a delay so the message won't be reprocessed immediately thus quickly exhausting it's max delivery count?
}

Answer 1

一种选择是创建一条新消息并将该消息提交到队列，但将 ScheduledEnqueueTimeUtc 设置为未来五分钟。

        [FunctionName("DelayMessage")]
        public static async Task DelayMessage(
            [ServiceBusTrigger("MyQueue", AccessRights.Listen, Connection = "MyConnection")]BrokeredMessage originalMessage,
            [ServiceBus("MyQueue", AccessRights.Send, Connection = "MyConnection")]IAsyncCollector<BrokeredMessage> newMessages,
            TraceWriter log)
        {
            //handle any kind of error scenerio

            var newMessage = originalMessage.Clone();

            newMessage.ScheduledEnqueueTimeUtc = DateTime.UtcNow.AddMinutes(5);

            await newMessages.AddAsync(newMessage);

        }

Answer 2

您现在可以使用 fixed delay retry，它于 2020 年 11 月左右添加到 Azure Functions（预览版）。

[FunctionName("MyFunction")]
[FixedDelayRetry(10, "00:05:00")]   // retries with a 5-minute delay
public static void Run([ServiceBusTrigger("someTopic", 
     "someSubscription", AccessRights.Manage, Connection = 
     "ServiceBusConnection")] BrokeredMessage message) 
{
     CallService(bodyOfBrokeredMessage); //service is down
}

Answer 3

正如 Josh 所说，您可以简单地克隆原始消息，设置计划的入队时间，发送克隆并完成原始消息。

好吧，很遗憾发送克隆和完成原始操作不是原子操作，所以如果处理过程在错误的时刻崩溃，我们再次看到原始的机会非常小。

另一个问题是克隆上的DeliveryCount 将始终为 1，因为这是一条全新的消息。所以我们可以无限地重新提交，并且永远不会死信这个消息。

幸运的是，这可以通过添加我们自己的重新提交计数作为消息的属性来解决：

[FunctionName("DelayMessage")]
public static async Task DelayMessage([ServiceBusTrigger("MyQueue", AccessRights.Listen, Connection = "MyConnection")]BrokeredMessage originalMessage,
            [ServiceBus("MyQueue", AccessRights.Send, Connection = "MyConnection")]IAsyncCollector<BrokeredMessage> newMessages,TraceWriter log)
{
     //handle any kind of error scenerio
     int resubmitCount = originalMessage.Properties.ContainsKey("ResubmitCount") ?  (int)originalMessage.Properties["ResubmitCount"] : 0;
     if (resubmitCount > 5)
     {
         Console.WriteLine("DEAD-LETTERING");
         originalMessage.DeadLetter("Too many retries", $"ResubmitCount is {resubmitCount}");
     }
     else
     {
         var newMessage = originalMessage.Clone();
         newMessage.ScheduledEnqueueTimeUtc = DateTime.UtcNow.AddMinutes(5);
         await newMessages.AddAsync(newMessage);
     }
}

更多详情可以参考这个article。

此外，在 LogicApp 中实现等待/重试/出列下一个模式非常容易，因为这种类型的流控制正是 LogicApps 的设计目的。请参考这个SO thread。