1.前言

上一篇我们讨论了云计算设计模式之管道和过滤器模式 了解了如何使用消息队列来协调多任务系统的执行。这一篇，我们继续深入讨论消息队列的使用，即消息队列中的消息是带有优先级别的，优先级别高的先处理，优先级别低的后处理。

2.概念

带优先级的消息的处理顺序与正常的顺序会有所不同。在向队列中添加消息时可用指定消息的优先级别,消息队列能够根据优先级重新调整处理顺序。

在不支持优先级的队列里，我们可以采用替代方案来解决这个问题,即采用多个队列来进行处理,每个队列有相应的消费者池,如下图所示.

这种模式下，消息消费者其实也可以共用一个池,当没有高优先级消息达到时,就处理普通级别的消息,当有高优先级的消息达到时,我们就处理高优先级消息.这种模式能够有效提升系统的性能.

在使用这种模式时以下内容值得考虑:

1）在解决方案的上下文中定义消息队列的优先级.

2）确定是否所有高优先级的消息都先于低优先级的消息处理.

3）在使用多队列,一个消费者的模式下,使用一个消费者池来侦听多个队列的方式比使用多个专有的消费者池的方式好.

4）需要设计一种记录队列中各个优先级的消息被处理的速度，确保消息处理的速度与预计的差距在可接受范围内.

5）如果要确保低优先级的消息一定会被处理,那么最好使用多消息队列的模式;

6）消息的优先级最好是由系统的逻辑来确定的.

7）在检查消息队列里面的消息的时候,可能涉及到费用的问题。通常在商用在消息进入队列,消息队列查询的过程中进行计费,并非所有的消息队列服务都是免费的.

8）最好动态决定由多少个消息消费者池.具体参考Autoscaling Guidance.

关于何时使用这种模式,官方说法如下:

* The system must handle multiple tasks that might have different priorities.

* Different users or tenants should be served with different priority.

3.Example

Azure 的消息队列通过排序自动实现了消息优先级服务,这意味着消息的优先级是由Azure自动决定的,使用者无法设置消息的优先级.为了解决这个问题,Azure提供了消息队列的主题（Topic）及订阅（subscriptions）服务,该服务能够实现消息过滤,支持消息优先级的实现.

在Azure Cloud Service中,调用这些消息队列都是通过Work Role来实现,下面的代码演示了基本的调用.

public class PriorityWorkerRole : RoleEntryPoint
{
  private QueueManager queueManager;
  ...

  public override void Run()
  {
    // Start listening for messages on the subscription.
    var subscriptionName = CloudConfigurationManager.GetSetting("SubscriptionName");
    this.queueManager.ReceiveMessages(subscriptionName, this.ProcessMessage);
    ...;
  }
  ...

  protected virtual async Task ProcessMessage(BrokeredMessage message)
  {
    // Simulating processing.
    await Task.Delay(TimeSpan.FromSeconds(2));
  }
}

高优先级消息的处理Work Role和低优先级消息的处理Work Role都实现ProcessMessage方法,下面代码演示了高优先级Role中的实现.

protected override async Task ProcessMessage(BrokeredMessage message)
{
  // Simulate message processing for High priority messages.
  await base.ProcessMessage(message);
  Trace.TraceInformation("High priority message processed by " +
    RoleEnvironment.CurrentRoleInstance.Id + " MessageId: " + message.MessageId);
}

在高优先级消息Role和低优先级消息Role中发布消息的做法如下:

// Send a low priority batch. 
var lowMessages = new List<BrokeredMessage>();

for (int i = 0; i < 10; i++)
{
  var message = new BrokeredMessage() { MessageId = Guid.NewGuid().ToString() };
  message.Properties["Priority"] = Priority.Low;
  lowMessages.Add(message);
}

this.queueManager.SendBatchAsync(lowMessages).Wait();
...

// Send a high priority batch.
var highMessages = new List<BrokeredMessage>();

for (int i = 0; i < 10; i++)
{
  var message = new BrokeredMessage() { MessageId = Guid.NewGuid().ToString() };
  message.Properties["Priority"] = Priority.High;
  highMessages.Add(message);
}

this.queueManager.SendBatchAsync(highMessages).Wait();

4.相关阅读

The following patterns and guidance may also be relevant when implementing this pattern:

• Asynchronous Messaging Primer. A consumer service processing a request may need to send a reply to the instance of the application that posted the request. The Asynchronous Messaging Primer provides more information on the strategies that can be used to implement request/response messaging.
• Competing Consumers Pattern. To increase the throughput of the queues, it’s possible to have multiple consumers that listen on the same queue, and process the tasks in parallel. These consumers will compete for messages, but only one should be able to process each message. The Competing Consumers pattern provides more information on the benefits and tradeoffs of implementing this approach.
• Throttling Pattern. You can implement throttling by using queues. Priority messaging can be used to ensure that requests from critical applications, or applications being run by high-value customers, are given precedence over requests from less important applications.
• Autoscaling Guidance. It may be possible to scale the size of the pool of consumer processes handling a queue depending on the length of the queue. This strategy can help to improve performance, especially for pools handling high priority messages.

• The article Priority Queue Pattern on the Cloud Design Pattern website.
• The article Enterprise Integration Patterns with Service Bus on Abhishek Lal’s blog.