如何确定毒队列消息的原因

问题描述：

我有一个天青的函数与QueueTrigger处理图像并输出blob以及表记录。如何确定毒队列消息的原因

有时在处理多个大图像时，我遇到了OutOfMemory异常，这可能导致队列项目放入毒性队列中。

有时会出现竞争条件，并且表记录被插入错误，因为已经有该分区键和行键的记录。

我可以在这些函数中自己解决这些问题，但是我的首选方法是处理这个消息，以便让有毒消息包含原因或导致项目被放入毒性队列的异常。这样我可以有另一个触发器监听毒素队列并在评估出错后采取相应措施。

这应该如何处理？我知道ErrorTrigger，它很好地得到这个异常，但是我没有看到我怎样才能将它与引起它的特定队列项联系起来。

我之前在ServiceBus Queue中通过将代码封装在try/catch和catch中，我得到了堆栈跟踪，并且自己将消息自己盲注了，并将堆栈跟踪作为消息包含进去。我不确定您是否可以明确地将存储队列消息放入毒性队列中。 –

我尝试了这种方法，但是我无法捕获表存储异常，因为它在函数返回后发生。我想它可以解决内存问题。 – Dexterity

答

在您的队列触发器失败后，无法更改毒性存储队列消息队列。

Azure功能使用Web Jobs SDK。直接使用Web Jobs SDK可以覆盖QueueProcessor.CopyMessageToPoisonQueueAsync。一些有用的链接：

https://github.com/Azure/azure-webjobs-sdk/issues/1204

https://github.com/Azure/azure-webjobs-sdk-samples/blob/master/BasicSamples/MiscOperations/CustomQueueProcessorFactory.cs

https://github.com/Azure/azure-webjobs-sdk/blob/ffae7c86ea87fd73748186bac2c38c5401b80f68/test/Microsoft.Azure.WebJobs.Host.EndToEndTests/AzureStorageEndToEndTests.cs

答

有跟踪异常和使用FunctionInvocation过滤器导致邮件的消息哈克的方式。

public static class QueueFunction 
{ 
    [FunctionName("QueueFunction")] 
    [TrackFailedMessages] 
    public static void Run([QueueTrigger("myqueue")]string message) 
    { 
     throw new Exception("OMG"); 
    } 
} 

public class TrackFailedMessages : FunctionInvocationFilterAttribute 
{ 
    private static readonly ConcurrentDictionary<Guid, object> QueueParamters = new ConcurrentDictionary<Guid, object>(); 
    public override Task OnExecutingAsync(FunctionExecutingContext executingContext, CancellationToken cancellationToken) 
    { 
     if (executingContext.Arguments.TryGetValue("message", out var data)) 
     { 
      QueueParamters.AddOrUpdate(executingContext.FunctionInstanceId, data, (id, obj) => data); 
     } 
     return Task.CompletedTask; 
    } 

    public override Task OnExecutedAsync(FunctionExecutedContext executedContext, CancellationToken cancellationToken) 
    { 
     if (executedContext.FunctionResult.Exception != null && 
      QueueParamters.TryGetValue(executedContext.FunctionInstanceId, out var message)) 
     { 
      executedContext.Logger.LogCritical(
       "The message {message} caused the exception {exception}", 
       message, 
       executedContext.FunctionResult.Exception.ToString()); 
     } 
     QueueParamters.TryRemove(executedContext.FunctionInstanceId, out var _); 
     return Task.CompletedTask; 
    } 
}

OnExecutinAsync将在调用函数时被调用。使用executableContext可以访问函数的参数。您可以搜索QueueTrigger参数并将该值存储在字典中，其中FunctionInstanceId是键。 FunctionInstanceId是每个调用的唯一标识。

在OnExecuted您可以检查函数引发和异常，并从字典中的消息，然后将其存储在别处（数据库，日志等）

我不知道这是否抓住了表存储异常...

如何确定毒队列消息的原因

相关推荐