代码之家 › 专栏 › 技术社区 › Oleg Sh

Azure函数:如何更好地实现队列消息重试的延迟

azure-queues azure-functions azure c#

Oleg Sh · 技术社区 · 7 年前

我的Azure函数应该监听队列中的消息,然后获取消息,尝试调用消息中带有值的外部服务,如果外部服务返回“确定”,那么我们必须将消息写入另一个队列(对于下一个Azure函数),如果返回“失败”,我们必须返回到当前队列,并在5分钟后由我们的Azure函数重试。S.如何实施?我用定时器做的,但解决方案不喜欢我:

    [FunctionName("FunctionOffice365VerificateDomain_and_AddService_and_GexMxRecord")]
    public async static Task Run([TimerTrigger("0 */5 * * * *")]TimerInfo myTimer,
        [Queue("domain-verificate-Office365-add-services-get-mx-record", Connection = "StorageConnectionString")]CloudQueue listenQueue,
        [Queue("domain-add-mx-record-to-registrator", Connection = "StorageConnectionString")]CloudQueue outputQueue,
        ILogger log)
    {
        while (true)
        {
            // do "invisible" message for next 30 sec
            var message = await listenQueue.GetMessageAsync();
            if (message != null)
            {
                DomainForRegistration domainForRegistration = JsonConvert.DeserializeObject<DomainForRegistration>(message.AsString);
                try
                {
                    await _office365DomainService.VerifyDomainAsync(domainForRegistration.DomainName);
                    // remove message
                    await listenQueue.DeleteMessageAsync(message);

                    await _office365DomainService.UpdateIndicateSupportedServicesDomainAsync(domainForRegistration.DomainName);

                    var mxRecord = await _office365DomainService.GetMxRecordForDomainAsync(domainForRegistration.DomainName);
                }
                catch (DomainVerificationRecordNotFoundException)
                {
                     // thrown when VerifyDomainAsync failed
                }
            }
            else
                break;
        }
    }

没有这些,如何更小心地做 while(true) ,但验证失败后是否超时?

1 回复 | 直到 7 年前

Jerry Liu Phantom 7 年前

同意@davidg,尝试使用队列触发器来实现您的目标。 W可以依靠 host setting 队列的

VisibilityTimeout是处理消息失败时重试间隔的时间。 MaxDequeueCount是将邮件移到病毒队列之前尝试处理邮件的次数。

{
    "version": "2.0",
    "extensions": {
        "queues": {
            "visibilityTimeout" : "00:05:00",
            "maxDequeueCount": 2,
        }
    }
}

这样,函数应该看起来像

public static async Task Run(
    [QueueTrigger("domain-verificate-Office365-add-services-get-mx-record")]string myQueueItem, ILogger log,
    [Queue("domain-add-mx-record-to-registrator", Connection = "StorageConnectionString")]IAsyncCollector<string> outputQueue
)
{
    // do stuff then output message
    await outputQueue.AddAsync(myQueueItem);
}

如果您不想将异常抛出给主机,我们可以转到 initialVisibilityDelay CloudQueue方法的。

指定从现在起消息将不可见的时间间隔

    public static async Task Run(
        [QueueTrigger("domain-verificate-Office365-add-services-get-mx-record")]string myQueueItem, ILogger log,
        [Queue("domain-add-mx-record-to-registrator", Connection = "StorageConnectionString")]IAsyncCollector<string> outputQueue,
        [Queue("domain-verificate-Office365-add-services-get-mx-record", Connection = "StorageConnectionString")]CloudQueue listenQueue
    )
    {

        try 
        {
            // do stuff then output message
            await outputQueue.AddAsync(myQueueItem);
        }
        catch(DomainVerificationRecordNotFoundException)
        {
            // add the message in current queue and can only be visible after 5 minutes
            await listenQueue.AddMessageAsync(new CloudQueueMessage(myQueueItem), null, TimeSpan.FromMinutes(5), null, null);
        }
    }