高效使用微软 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({1}
             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 读者交流群）。