如何防止Azure webjob同时多次处理相同的消息

Question

我有一个Azure WebJob项目,我在我的开发机器上本地运行.它正在侦听Azure Service Bus消息队列.什么都没有像主题,只是最基本的消息队列.

它多次接收/处理相同的消息,在收到消息时立即启动两次,然后在处理消息时间歇性地启动.

问题:

• 为什么我立即多次收到同样的信息？它似乎是在应用PeekLock之前重新获取的？
• 为什么即使它仍在处理中,该消息又被重新接收？我可以设置PeekLock持续时间,或以某种方式锁定消息只处理一次
• 如何确保队列中的每条消息只处理一次？
• 我希望能够一次处理多个消息,多次不同的消息,所以将MaxConcurrentCalls设置为1似乎不是我的答案,或者我误解了该属性？

我正在使用异步函数,简单的注入器和自定义JobActivator,因此我的函数签名是:而不是静态void方法:

public async Task ProcessQueueMessage([ServiceBusTrigger("AnyQueue")] MediaEncoderQueueItem message, TextWriter log) {...}

在工作中,它正在移动blob服务上的一些文件,并从媒体服务调用(和等待)媒体编码器.因此,虽然Web作业本身没有进行大量处理,但是需要相当长的时间(对于某些文件,这需要15分钟).

该应用程序正在启动,当我向队列发布消息时,它会响应.但是,一旦收到消息,它就会多次收到消息:

Executing: 'Functions.ProcessQueueMessage' - Reason: 'New ServiceBus message detected on 'MyQueue'.'
Executing: 'Functions.ProcessQueueMessage' - Reason: 'New ServiceBus message detected on 'MyQueue'.'

此外,当任务正在运行时(我看到媒体服务功能的输出),它将从队列中获得另一个"副本".

最后,在任务完成后,它仍然间歇性地处理相同的消息.

Answer 1

我怀疑发生的事情如下:最多DurationLock可以是5分钟.如果在5分钟内完成消息处理,则消息被标记为已完成并从代理中删除.否则,如果处理时间超过5分钟(我们丢失了对消息的锁定),将重新显示消息,并将再次消耗.您可以通过查看DeliveryCount邮件来验证.

要解决此问题,您可以在即将到期之前使用更新消息锁定BrokeredMessage.RenewLockAsync().