azure - Azure Functions 中的服务总线死信队列处理

Question

我的环境。

• dotnet 6
• Azure Functions(点网隔离)
• Azure 服务总线

现在我有两个带有 ServiceBusTrigger 的函数。一个处理主队列消息，另一个处理死信队列消息。

场景

• 当主队列消息处理失败(两次)时，消息将移至死信队列。
• 现在它会触发死信队列消息并开始重试。这里似乎是无限循环。如果消息在 DLQ 级别失败，如何将消息移至另一个队列。有什么办法可以重新将消息提交到队列中。

更新1

• 场景是处理该消息的消费者出现问题。
• 现在，当消息首先进入主队列，然后由消费者处理时，由于与消费者的某些依赖性问题，出现了问题。因此，经过一些重试后，消息将进入 DLQ。
• 现在DLQ使用相同的业务代码来处理消息，但它来自不同的函数。由于依赖关系有问题，因此消息将再次失败。现在在 DLQ 中它处理无限。我想将此消息移至其他队列，并从那里需要一些手动干预来进一步处理该消息。
• 在高流量应用中，在做出上述决定之前，队列/DLQ 中可能有很多消息。

Answer 1

从 DLQ 获取消息的进程不应该经常失败。 DLQ 中的消息本身进入循环这一事实表明，这里确实出了问题。

您能否详细说明您的流程正在做什么以及为什么失败？您是否从主队列触发与从 DLQ 触发相同的功能？那是不明智的。如果您可以查看失败的根本原因，那么在 DLQ 中需要处理的消息就会减少。

这样想吧。您是否需要额外的队列来重试，然后将其推送到另一个队列？在最终将其移至死信队列之前，您还需要多少个队列？那么你会如何处理死信队列呢？

无论如何，服务总线不会自动将消息从 DLQ 移动到另一个队列。但如果确实需要，您可以触发自己的进程将其写入另一个队列。服务总线触发的 Azure 函数非常适合完成此类工作。

要打破 DLQ 的无限循环，请添加对 Brokered 属性中的递送计数值的检查。如果该值大于某个阈值，则将该消息写入另一个队列。每次消息被拾取时，传递计数都会增加。

下面显示了 .NET 中的一个示例。我正在模拟失败来说明。

using System;
using System.Threading.Tasks;
using Microsoft.Toolkit.Services;

public static void Run(string myQueueItem, Int32 deliveryCount, ILogger log)
{
    log.LogInformation($"C# ServiceBus queue trigger function processed message: {myQueueItem}");
    log.LogInformation($"Delivery count : {deliveryCount}");
    if (deliveryCount < 3)
    {
        log.LogInformation("Simulating failure");
        throw new TooManyRequestsException();
    }
    else
    {
        log.LogInformation($"Delivery count is now {deliveryCount}. Time to write this message into another queue.");
    } 
}

在第三次重试时，我想将消息发送到另一个队列。向我的服务总线发送测试消息后，我得到以下输出。

2022-02-22T03:05:35.281 [Information] Executing 'Functions.ServiceBusQueueTrigger1' (Reason='(null)', Id=eed59015-f100-42b4-b0bf-57a9a6ab226f)
2022-02-22T03:05:35.281 [Information] Trigger Details: MessageId: 86f8250e82fb42deb88e6b91244ca2b7, SequenceNumber: 3, DeliveryCount: 1, EnqueuedTimeUtc: 2022-02-22T03:05:35.2980000Z, LockedUntilUtc: 2022-02-22T03:06:05.2980000Z, SessionId: (null)
2022-02-22T03:05:35.531 [Information] C# ServiceBus queue trigger function processed message: my new message
2022-02-22T03:05:35.531 [Information] Delivery count : 1
2022-02-22T03:05:35.532 [Information] Simulating failure
2022-02-22T03:05:35.557 [Error] Executed 'Functions.ServiceBusQueueTrigger1' (Failed, Id=eed59015-f100-42b4-b0bf-57a9a6ab226f, Duration=253ms)Exception of type 'Microsoft.Toolkit.Services.TooManyRequestsException' was thrown.
2022-02-22T03:05:35.644 [Information] Executing 'Functions.ServiceBusQueueTrigger1' (Reason='(null)', Id=78d2df93-11f6-412b-9ae7-880ac3bf4213)
2022-02-22T03:05:35.645 [Information] Trigger Details: MessageId: 86f8250e82fb42deb88e6b91244ca2b7, SequenceNumber: 3, DeliveryCount: 2, EnqueuedTimeUtc: 2022-02-22T03:05:35.2980000Z, LockedUntilUtc: 2022-02-22T03:06:05.6730000Z, SessionId: (null)
2022-02-22T03:05:35.645 [Information] C# ServiceBus queue trigger function processed message: my new message
2022-02-22T03:05:35.646 [Information] Delivery count : 2
2022-02-22T03:05:35.646 [Information] Simulating failure
2022-02-22T03:05:35.674 [Error] Executed 'Functions.ServiceBusQueueTrigger1' (Failed, Id=78d2df93-11f6-412b-9ae7-880ac3bf4213, Duration=2ms)Exception of type 'Microsoft.Toolkit.Services.TooManyRequestsException' was thrown.
2022-02-22T03:05:35.782 [Information] Executing 'Functions.ServiceBusQueueTrigger1' (Reason='(null)', Id=b856141d-d891-4edb-a910-7e53b40b96a0)
2022-02-22T03:05:35.783 [Information] Trigger Details: MessageId: 86f8250e82fb42deb88e6b91244ca2b7, SequenceNumber: 3, DeliveryCount: 3, EnqueuedTimeUtc: 2022-02-22T03:05:35.2980000Z, LockedUntilUtc: 2022-02-22T03:06:05.8130000Z, SessionId: (null)
2022-02-22T03:05:35.783 [Information] C# ServiceBus queue trigger function processed message: my new message
2022-02-22T03:05:35.784 [Information] Delivery count : 3
2022-02-22T03:05:35.784 [Information] Delivery count is now 3. Time to write this message into another queue.**
2022-02-22T03:05:35.784 [Information] Executed 'Functions.ServiceBusQueueTrigger1' (Succeeded, Id=b856141d-d891-4edb-a910-7e53b40b96a0, Duration=2ms)