等待Service Bus OnMessage处理的消息完成

Question

我正在使用Azure Service Bus SubscriptionClient.OnMessage方法; 配置为同时处理多达5个消息.

在代码中,我需要等待所有消息完成处理才能继续(正确关闭Azure Worker角色).我该怎么做呢？

SubscriptionClient.Close()会阻塞,直到所有消息都完成处理？

Answer 1

在SubscriptionClient或QueueClient上调用Close不会阻止.据我所知,Calling Close会立即关闭实体.我只是使用Windows Azure SDK 2.0附带的Worker Role With Service Bus Queue项目模板快速测试.我在消息进程操作中添加了一个线程休眠几秒钟,然后在它运行时关闭该角色.我看到在消息处理它们的线程睡眠时调用了Close方法,但它肯定没有等待for消息处理完成,角色简单关闭了.

为了妥善处理这个问题,您需要执行我们在处理处理消息(Service Bus,Azure存储队列或其他任何内容)的任何工作者角色时所做的相同操作:跟踪正在处理的内容并在其处理时关闭完成了.有几种方法可以解决这个问题,但由于涉及多个线程,所有这些方法都是手动的并且在这种情况下变得混乱.

鉴于OnMessage的工作方式,您需要在操作中添加一些内容,以查看角色是否已被告知关闭,如果是,则不进行任何处理.问题是,当执行OnMessage操作时,它已经有了一条消息.您可能需要放弃该消息但不要退出OnMessage操作,否则如果队列中有消息,它将继续收到消息.您不能简单地放弃消息并让执行离开操作,因为系统将被传递另一条消息(可能是同一条消息),并且执行此操作的多个线程可能会导致消息得到太多的出队计数并且会被删除.此外,您不能在SubscriptionClient或QueueClient上调用Close,这将在内部停止接收循环,因为一旦您调用close,任何未完成的消息处理将在.Complete,.Abandon等被调用时抛出异常.消息,因为消息实体现在已关闭.这意味着您无法轻松停止传入的消息.

这里的主要问题是因为您正在使用OnMessage并通过在OnMessageOptions上设置MaxConcurrentCalls来设置并发消息处理,这意味着启动和管理线程的代码隐藏在QueueClient和SubscriptionClient中并且您没有控制权在那之上.您没有办法减少线程数,或单独停止线程等.您需要创建一种方法将OnMessage操作线程置于他们知道系统被告知的状态关闭然后完成他们的消息而不退出操作,以便不再为他们分配新消息.这意味着您可能还需要将MessageOptions设置为不使用自动完成,并在OnMessage操作中手动调用complete.

必须完成所有这些操作可能会严重降低使用OnMessage帮助程序的实际好处.在幕后OnMessage只是设置一个循环调用接收默认超时和将消息处理到另一个线程来执行操作(松散描述).因此,使用OnMessage方法所得到的就是不必自己编写该处理程序,但是您遇到的问题是因为您没有自己编写该处理程序而您无法控制这些线程.第二十二条军规.如果你真的需要优雅地停止,你可能想要离开OnMessage方法,用线程编写你自己的Receive循环,并在主循环内停止接收新消息并等待所有工人结束.

一个选项,特别是如果消息是幂等的(这意味着处理它们不止一次产生相同的结果......无论如何你都应该注意)然后如果它们在中间处理中停止,它们将简单地重新出现在要处理的队列上之后的另一个例子.如果工作本身不是资源密集型而且操作是幂等的,那么这确实是一种选择.与实例可能由于硬件故障或其他问题而失败时没有什么不同.当然,它不是优雅或优雅,但它肯定消除了我提到的所有复杂性,并且仍然是由于其他故障可能发生的事情.

请注意,当告知实例关闭时,将调用OnStop.你有5分钟可以延迟这个,直到布料关闭它,所以如果你的消息需要超过五分钟的时间来处理它,如果你试图优雅地关闭也没关系,有些将会被切断在处理过程中