优雅地推迟QueueTrigger

Question

我目前正在使用WebJobs软件开发工具包来使用队列中发出的消息。

我的方法作为一个带有Microsoft.Azure.WebJobs.QueueTrigger(...)的参数属性，并被正确触发。在某些情况下，该方法可以处理消息，但有时，我更希望它在关键资源可用之前拒绝该消息。

在这种情况下，我尝试抛出一个异常，但与引用所说的相反，队列触发器立即再次触发(显然不等待租用时间)。

有没有一种方法可以优雅地推迟消息处理？冻结等待关键资源的线程是否安全？

任何提示都将不胜感激。

Answer 1

我不认为你可以在当前版本中推迟消息。

可能的解决方法

你可以用延迟重新添加相同的消息，为了避免重复，你可以将你的MaxDequeueCount设置为1，这将在异常后将失败的消息直接发送到中毒队列：

        JobHostConfiguration configuration = new JobHostConfiguration();
        configuration.Queues.MaxDequeueCount = 1;

和message processor -重新添加带有延迟和抛出异常的消息：

    public static void ProcessMessage([QueueTrigger("resource-heavy-queue")] string message, [Queue("resource-heavy-queue")]  CloudQueue originalQueue)
    {
        if ( /*Resource unavaliable*/)
        {
            var messageToReAdd = new CloudQueueMessage(message);
            originalQueue.AddMessage(messageToReAdd, null, TimeSpan.FromSeconds(10));
            throw new ResourcesNotAvailableException();
        }
    }

通过这种方式，您可以为您的资源实现某种回退策略。不幸的是，您必须手动处理一些问题：

• 处理有害消息-如果您不断重新添加相同的消息，您可能会陷入无限循环，因此您必须扩展您的消息模型以携带NumberOfRetries，并在每次重新添加时递增它。
• Message Id和InsertionTime在每次重新添加后都会不同，因此您不能依赖它们。

Answer 2

抱歉，无法推迟触发。您需要的是一个多触发器(当消息和资源可用时触发)，这是Azure WebJobs SDK所没有的

这里有一些变通方法。如果队列中的所有消息都需要关键资源，那么可以冻结线程--你正在实现一个信号量--否则，调度会变得很棘手，因为如果你达到了并行运行的函数的最大数量，SDK就不会处理新的消息。

我要做的不是在队列消息上触发，而是在关键资源上触发。当资源可用时，它会将一条消息放入另一个队列中，然后检查是否有需要该资源的消息需要处理。

Answer 3

在这种情况下，我要做的是为触发我的同一队列创建一个ICollector，如下所示

public static async Task HandleMessagesAsync([ServiceBusTrigger("%QueueName%")] BrokeredMessage message, [ServiceBus("%QueueName%")]ICollector<BrokeredMessage> queue, TextWriter logger)

( %QueueName%表示法使SDK从app.config中获取值)

然后在处理程序中，我做了一些类似的事情

if (needToWait)
{
    var delayedMessage = new BrokeredMessage(originalMessageBody) { Label = originalLabel, MessageId = originalMessageId, ScheduledEnqueueTimeUtc = DateTime.UtcNow + this.ExecutionDelay };
    queue.Add(delayedMessage);
    return;
}

这样，当前消息就完成了，但具有相同属性的新消息被安排在定义的超时之后发送。

在这个程序的制作过程中，没有线程被阻塞。