高效使用微软Azure服务总线的消息功能

无论是应用系统之间，或者分布式系统的不同组件之间，通常需要一种快速、安全、可靠的方式来彼此通信。以前，你会想到基于MSMQ或RabbitMQ之类消息中间件来实现，自己搭建和维护的同时，确保可用性、扩展性和性能等符合系统设计要求。

Windows Azure服务总线（Service Bus）提供了一套安全且高可用的基于云服务的托管消息传递基础设施，以实现广泛通信、大范围事件分布、命名和服务发布。这篇文章会介绍Azure服务总线消息功能的基本概念、适用场景、以及一些实践经验。

Azure服务总线中的队列和主题/订阅

Windows Azure支持两种可靠的处理异步消息的队列机制：

• 队列（Queue）：一种简单有保障的先进先出（FIFO）消息传递方式。也就是说，消息通常按其加入队列的临时顺序由接收方接收和处理，并且每条消息只能由一个消息使用者接收和处理。
• 主题/订阅(Topic/Subscriptions)：提供一对多的通信形式，将消息传递到多个订阅。订阅上可以设置筛选规则来限制要接收的消息。先将消息发送到主题，消息可能被多个订阅接收，然后，接收方在订阅上接收消息。也就是说，消息可能被多个订阅上的不同接收方接收。

解决的问题和适用场景

假设这样一个场景：系统/组件A（简称A）向系统/组件B（简称B ）发送消息。但如果B由于设备或网络原因而无法访问时，对A会造成什么影响呢？显然，在下图所显示的直连模式下（A调用B的服务接口），A功能会因为B不可用而无法正常提供服务。

为了降低B不可用对A造成的影响，可能你会考虑通过增加B的实例数，以提高B的可用性。而且，实际情况中往往A也是多实例的。

显然，你会不希望看到上图的情况发生在自己的系统设计中，因为这种紧耦合的方式并不宜于系统的维护和扩展。因此，可以考虑在A和B的实例间使用负载均衡来解决，这样一来，无论B有多少实例，对A来说只需要配置一个访问地址。

虽然负载均衡提高了A和B之间的松耦合性，但新的问题来了。当A在遭遇到流量高峰时，高流量带来的压力也会传递到B上来。如果B的处理能力不足，B的响应速度会变慢，甚至瘫痪。可见，负载均衡并未提供到百分百的松耦合。

如果你也遇到了上述类似的问题，那不妨在应用设计中考虑使用Azure服务总线的队列或主题/订阅（如下图所示）。

在基于队列或主题/订阅的异步消息处理机制下，系统会具备如下特点：

• 消息的持久化: A发送的消息会保持在队列中，即使B不可用，消息不会丢失。
• 负载平衡：无论A的负载如何变化，只会影响队列中消息的数量，B始终按照自己的处理能力处理消息。
• 负载均衡：通过监控队列中消息数量，可更有效的控制A和B的实例数。比如，消息数量累计较多，意味B的处理能力不够，可以通过增加B实例的方式提供消息的处理能力。反之，可减少A或B的实例数。

Windows Azure服务总线的队列或主题/订阅的适用场景其实很多，主要包括两类：

1. 系统组件间的异步消息传递，将耗时的过程分离出来。例如，视频网站接收到视频文件后，需要编码以支持不同终端设备播放，而编码过程通常比较慢，这时，接收模块和编码模块间可以考虑使用队列来实现异步处理。
2. 系统间的异步消息传递，将耦合降到最低。例如，一个微信平台，接收用户的输入，从第三方系统接收结果，再发送给用户。微信后台系统接收到用户输入后，可以将消息通过主题/订阅发送给第三方系统(通过设置订阅上的筛选规则来过滤哪些消息发送给哪些订阅)；第三方系统从绑定的订阅中获取消息并处理，然后将结果发送到一个新的队列中；微信后台系统从队列中接收结果，再返回给用户。
3. 设备与系统间的消息传递，起到物联网网关作用。例如，智能家居中的智能设备（空气净化器、烟雾传感器等）不断将设备状态发送到云端，用户通过手机上的应用可以远程控制设备。队列、主题/订阅在设备与云端系统间担当网关，采集设备状态，分发控制指令等。

一些经验分享

• 了解Azure存储队列和服务总线消息的区别

  Azure存储队列: 提供了可靠的异步消息传递和简单的REST接口。但不提供排序保障，不支持复杂的消息功能和传递模式。

  Azure服务总线消息: 相比于存储队列，基于更广泛的“中转消息传送”基础结构而构建，支持REST/AMQP协议，保障排序（先进先出），提供多种消息功能（例如，去重/事务/会话等），支持不同的消息传递模式（例如，主题/订阅、自动转发等）。

  关于详细的对比分析，可以参考微软MSDN上的文章“ Azure 队列和 Service Bus 队列 - 比较与对照 ”。一些常见功能的对照如下图所示：

  功能	存储队列	服务总线消息
  协议	REST	REST, AMQP
  消息序列化	字符串 字节数组	可序列化对象 流
  最大消息大小	64 KB	256 KB
  最大消息生存期	7天	无限
  接收模式	Get and Delete	Receive and Delete Peak-Lock
  发布-订阅 （主题）		支持
  邮件到期时间	支持	支持
  死信队列		支持
  会话		支持
  消息延迟		支持
  重复检测		支持
  事务		支持
  计费	存储、 事务数	消息吞吐量、连接数

• 留意Azure服务总线的天花板在哪

  支持多租户的云服务不可能为单一租户提供无尽的资源。为了确保单一租户在使用资源时不影响其他租户，Azure服务总线为消息传递设置了配额，如下图所示：

  每个订阅下的最大命名空间数量	100
  队列/主题的最大大小	5 GB。 如果启动分片（Partition），80GB
  每个命名空间的最大队列/主题数量	10000
  每个主题的最大订阅数量	2000
  每个主题的SQL 筛选器最大数量	2000
  每个主题的相关性筛选器最大数量	100000
  每个命名空间的最大并发连接数	NetMessaging: 1,000 AMQP: 5,000 REST操作不计入
  队列/主题/订阅实体上的最大并发连接数	依赖于上值
  队列/主题/订阅实体上的最大并发接收请求数	5000

  当系统的设计要求超过单一队列/主题/订阅实体、或者单一命名空间、甚至单个订阅时，可以采用组合方式来解决。

  上述阈值将来可能会发生变化，请参考MSDN上的文章“ 服务总线配额 ”，获取最新信息。

• 清楚如何选择队列还是主题/订阅

  队列和主题/订阅最大的区别在于：队列中的消息只能被一个接受方接收；而主题/订阅允许消息被多个接受方接收。也就是说，主题/订阅具有队列全部的功能，但支持具有过滤规则的订阅功能。

  至少一次（At-least-one）的消息传递

  由于网络问题或其他不可预知的短暂性异常出现，在发送消息到队列或主题的过程中，可能出现失败。为应对这一问题，建议在消息发送方采用重试（Retry）机制。C#示例代码如下：

  public static async Task SendWithRetryAsyn(  Func<BrokeredMessage, AsyncCallback, object, IAsyncResult> beginSend,  Action<IAsyncResult> endSend,  Func<BrokeredMessage> getMessage,  Action<BrokeredMessage> setProperties = null,  int maxRetries = DefaultMaxRetries,  int retryBackoff = DefaultRetryBackoff) {  int retries = 0;  if (maxRetries < 0)  {   maxRetries = DefaultMaxRetries;  }  if (retryBackoff < 0)  {   retryBackoff = DefaultRetryBackoff;  }  bool sent = false;  while (!sent)  {   try   {    // acquire the message from the delegate    var brokeredMessage = getMessage();    if (setProperties != null)    {     // call the delegate to overrid ethe properties     setProperties(brokeredMessage);    }    // send the message    await Task.Factory.FromAsync(beginSend, endSend, brokeredMessage, null, TaskCreationOptions.None);    sent = true;   }   catch (MessagingException exception)   {    // for transient errors go to the Retry backoff wait    if (exception.IsTransient)    {     if (++retries > maxRetries)     {      throw;     }     goto Retry;    }    // otherwise throw out    throw;   }   continue;   Retry:   // wait for 3^retries * retryBackoff and then retry   await Task.Delay((int) Math.Pow(3, retries)*retryBackoff);   continue;  } } 

• 最多一次（At-most-one）的消息传递

  消息发送方如果处理不当，可能会将同一消息向同一个队列/主题中发送多次。你可以启动重复检测（Duplicate Detection）,让队列/主题自动将重复的消息删除。

  1) 队列/主题的重复检测检测的是消息中的MessageId 属性。当发现有重复的MessageId 出现时，会被认为是重复消息。所以，你可以利用MessageId存储一些有意义的信息，比如订单号。

  2) 谨慎设置队列/主题上的DuplicateDetectionHistoryTimeWindow属性。默认是10 分钟，不超过7天。该属性决定了队列/主题从多久的时间范围内所接收的消息中排查重复。时间设置的越久，意味着队列/主题需要更多的存储空间存储MessageId，这将占用正常的存储空间。

• 认清消息接收方式Receive and Delete和Peak-Lock的区别

  如果消息接收方使用Receive and Delete方式获取消息，那意味着接收端接收到消息的同时，消息会马上从队列/主题上删除。但可能会造成消息的丢失，比如消息被接收后，在还没有处理完时系统宕机了。

  为了避免消息丢失，建议采用Peak-Lock方式。但Peek-Lock有可能让消息被处理多次。例如，消息接收方收到一个消息后，需要较长时间来处理消息，超过了Lock Timeout时间（默认是60秒）还未向队列/主题发送完成（Complete）指令。这种情况下，消息会被解锁，其它消息接收方可以再次看到该消息，重复接收并处理。

  消息接收方在设计时应该充分考虑如何避免重复处理。例如，合理设置Lock Timeout时间(大于消息处理平均时间)；利用数据库或存储，记录消息处理状态，每次消息处理前，检查其状态，以避免重复处理。

• 一组消息，仅由一个消息接收方处理

  由于队列/主题的消息大小限制（256K），有时候需要将一个大的消息拆分成若干的小的消息。而为了能够完整的处理大消息，这些小的消息需要能够被同一个消息接收方接收，而后拼装成大消息。

  队列/主题的会话功能可以实现一组消息仅被一个消息接收方接收。

• 请求/响应的消息交换模式

  队列/主题普遍运用在单向（One way）的消息交换模式，但是否支持请求/响应模式呢？是的，可以利用启动了会话功能的两个队列/订阅来实现。

  

  基本的工作流程是：

  1) A将一个设置了ReplyToSessionID属性的消息发给Request队列，并且等待从Response队列中接收SessionID等于ReplyToSessionID的消息；

  2) B接收到Request队列中的请求消息，然后处理；

  3) B处理完成后，生成一个新的响应消息，并将其SessionID属性的值设为请求消息中ReplyToSessionID属性的值。 然后，将响应消息发送到Response队列中；

  4) 最后，A从Response队列中获得响应消息。完成。

• 横向扩展，增加消息吞吐量

  1) 自定义路由规则连接多个队列或订阅

  如前面所提到的，当吞吐量超出单一队列/主题的配额时，可以创建多个队列/主题，通过自定义的路由机制转发消息到不同的队列/主题中。

  

  2) 利用自动转发（Auto-forwarding）连接多个队列或订阅

  

  队列/订阅的自动转发（Auto-forwarding）功能允许你将队列或订阅中的消息自动转发到同一个命名空间下的另外一个队列或主题中。利用这种技术，可以扩展单一队列或订阅，改善吞吐量。

• 重用消息工厂（MessageFactory）

  C#中，每实例化一个MessageFactory 类，都会建立一个与Azure服务总线的TCP连接。在同一个消息发送方或接收方，建议采用单例模式，可以有效的减少与服务总线的并发连接数，并节省每次创建MessageFactory实例带来的性能开销。

• 一个不错的Azure服务总线管理工具，叫ServiceBusExplorer

  来自微软的Paolo Salvatori开发了这一工具，下载地址是 MSDN 。通过ServiceBusExplorer工具，开发者或运维人员可以方便的管理、监控和测试服务总线命名空间下的队列、主题和订阅。Paolo SalvatoriPaolo SalvatoriPaolo Salvatori

• 中国Azure和国外Azure的服务总线命名空间的后缀地址是不同的

  中国Azure的后缀地址是.servicebus.chinacloudapi.cn

  国外Azure的后缀地址是.servicebus.windows.net

• 监控消息吞吐量

  在消息发送方或接收方，可以在代码中自定义性能计数器来记录每秒消息的发送/接收数，可以利用Perfmon 工具实时查看。记录消息接收数的示例代码如下：

  private static void OnMessageArrived(BrokeredMessage message) {  var custmsg = message.GetBody<CustomMessage>();  System.Diagnostics.PerformanceCounter perfCounterReceived =   new System.Diagnostics.PerformanceCounter("SBMulticastDemo",     "Message Receive/Sec",     RoleEnvironment.CurrentRoleInstance.Id,     false);  if (perfCounterReceived != null)  {   perfCounterReceived.ReadOnly = false;   perfCounterReceived.IncrementBy(1);  } } 

参考文档

https://msdn.microsoft.com/zh-cn/library/hh367516.aspx

https://msdn.microsoft.com/zh-cn/library/hh767287(VS.103).aspx

https://msdn.microsoft.com/en-us/library/azure/ee732538.aspx

感谢杨赛对本文的审校。

给InfoQ中文站投稿或者参与内容翻译工作，请邮件至editors@cn.infoq.com。也欢迎大家通过新浪微博（@InfoQ，@丁晓昀），微信（微信号： InfoQChina ）关注我们，并与我们的编辑和其他读者朋友交流（欢迎加入InfoQ读者交流群 ）。