高级消息队列协议在大数据传输中问题及解决

张逸凡　于志安

摘要：高级消息队列协议是针对面向消息中间件的开放式标准的应用层协议规范，它面向消息并以队列存储，支持点对点及pub-sub的路由，具有可靠性和安全性。作为线路层协议，高级消息队列协议允许来自不同用户的消息生产者和消费者实现真正的互操作扩展，就如同SMTP、HTTP、FTP等协议采用的方式一样。该文以高级消息队列协议的消息机制为基础，探索该协议于流式信息（或文件）传输的可行性，和对其拓展的展望。

关键词：AMQP；消息队列；协议；消息中间件；消息路由；协议缓冲

中图分类号：TP393 文献标识码：A 文章编号：1009-3044（2014）01-0047-04

高级消息队列协议（Advanced Message Queuing Protocol），以下简称AMQP协议，是一个异步消息传递所使用的应用层协议规范。它的特点是面向消息，存储队列，支持点对点及pub-sub的路由，具有可靠性和安全性[1][3]。

作为线路层协议，AMQP允许来自不同用户的消息生产者和消费者实现真正的互操作扩展，就如同SMTP、HTTP、FTP等协议采用的方式一样。而此前对于消息中间件的标准化努力则集中在API层面上（比如JMS），且没有提供互操作性的途径。不同于JMS的仅仅定义API，AMQP是一个线路层的协议——它描述了通过网络传输的字节流的数据格式。因此，遵从这个协议的任何语言编写的工具均可以操作AMQP消息[4]。

当前关于AMQP的实现对象大都是企业级的内部消息应用（如JP Morgan）或是一些封闭系统的中间消息层（如red hat），它们主要面向的都是以轻量文本消息为介质的对象（可能也是出于稳定性的考虑）。但在大数据应用背景下，使用消息系统进行大流量的传输过程存在数据单元离散化产生的的传输性能瓶颈、并发输出和消息数据承载格式受限等问题，该文试图与AMQP结合使用Google的Protocol buffer，通过其提供的先进的可编程编解码方式来改善其传输性能并解决相应的问题。

1 AMQP协议

1.1 起源

AMQP由JP Morgan自2004年中期至2006年中期独自开发。IMatix公司也开发了C/C++和java的实现。两家公司将这个协议文档化并且规范了互操作性，发起成立专门的工作组，成员有JP Morgan Chase、Red Hat、 Cisco Systems、 TWIST、 IONA、 iMatix等。AMQP作为规范发布于2006年1月20日，工作组在设计时考虑了性能和可交互性，为基于队列的消息机制订立了一个协议和模型。AMQP协议的目标是普及消息中间件并跨越不同技术门槛，在各种语言或者各种操作系统之间提供真正的可交互性。出现了许多基于AMQP的软件商及其产品，表1列出了部份AMQP的软件商和其产品，具有很好的交互性，比如，重载了AMQP的JMS API能够很容易地与.net客户端以及任何其他语言或其他借助AMQP通讯的平台进行交互。

1.2协议概念结构

AMQP协议由是关键实体和语义表示的逻辑框架，主要有Broker（服务器）地址、用户信息、Exchange（信息交换器）、Exchange-Type（交换器类型）、Queue（消息队列）、Routing-Key（路由关键词）和Message（消息体）等。其中：

1） Broker是指AMQP客户端所要连接的服务器实例，是消息协议的核心部分。

2） 用户信息包含了用户名、密码等验证信息，来通过AMQP的用户认证机制并接入Broker，但是也可以不经认证直接接入。

3） Exchange是消息送达的实体，在broker中扮演消息路由结点的角色，除了在一对一这样的最基本的情况外，是消息流的必经之地。

4） Queue是接受消息的实体，具有名称和属性，但没有类型。客户端可以订阅队列以便让Broker递送某个消息队列的内容到该客户端。另外，客户端也可自动到队列取他所感兴趣的消息。消息确保以自己被送往队列时顺序分发，但在进行某些重新路由的操作如失效处理时，其顺序则不能保证。（注：AMQP 1.0中，队列将替代交换器。）

5） Routing-Key位于消息头，它的使用与否取决于交换器的类型。它的作用是作为用来与交换器匹配的关键词。

6） Exchange-Type定义了交换器的工作模式，一般分为Direct、Fan-Out、Topic三种方式。

1.3 消息路由模型（定义文字不清晰）

理解消息如何传递到队列的关键在于交换器的类型与路由关键词之间的关系。当一个消息的路由关键词与绑定中的模式匹配时，交换器会把该消息的拷贝送达队列。如何进行匹配仅依赖于交换器的类型：

① Direct型：消息的路由关键词与绑定相同。

② Fan-Out型：总是匹配，即使绑定无关键词。

③ Topic型：匹配路由关键词属性，字符串的各个部分以'.'分隔。可包含两个特殊字符：'*'表示单个任意词，'#'表示0个或多个词，例如 *.stock.# 匹配usd.stock和eur.stock.db，但不匹配stock.nasdaq。

根据以上元素组合可以组成一些典型的消息路由模型。

灵活地组合AMQP结构就可以很方便的在跨平台和跨语言的基础上搭建“消息广播”，“即时通信”，“电子邮件”等这类常见消息应用的架构，极大地简化了工作量。当然，这也正是该开源协议的设计初衷。

2 基于AMQP協议的大数据传输问题及原因

当AMQP协议用于互联网应用时，就会遭遇到大数据传输问题和不同的数据承载格式。以常用的电邮系统为例，现代电子邮件已经不再仅仅是一组字符串的信息段的组合了，可能会包括图片或是网页剪辑，尤其是大容量附件；另外，组合使用多种数据格式也非常普遍，图片、视频流和音频流已是主流应用的标准配置。

2.1存在问题（没点出问题，分析部份应作为2.2中，作为分析的一部分）

AMQP原本设计的目上开看，是为了实现一个通用的标准让网络具有消息中间件的能力，从而降低企业和系统集成上的开销。其使用对象主要在于系统级别用户，但对于单个系统节点或是企业内部节点网络而言，还远远达不到AMQP所提供的性能支持上限。所以，将AMQP的实现应用于较大信息量的交换也是有一定前景的，但这还需要进行进一步的实践。这里以100k文本数据量为例，模拟分段传输100～1000个该数据量的数据段并计算时间。

圖5 数据传输测试统计

由图5可以看出，基本上时间和数据量基本上是以线性的关系增长的。共100m文件传输所花的时间也仅稍高于本地硬盘数据传输的开销，所以在分开请求数据的代价比较小。

2.2 主要原因

可见一般情况下AMQP的主要瓶颈在于输入/输出端的响应时间。那是否是说这样的话那分包数越小对性能就越好呢？当然不是，这仅仅是理论情况。在实际运用中，数据的完整性也是需要相当考量的问题。若是一个近几百兆的数据段在传输的过程中发生了中断或是缺失这样的情况，数据分包自然就不是越少越好了。

需要在两者之间找到一种平衡，接下来会用一种数据流式化的方法来解决这一问题。

3 一种基于AMQP协议的大数据传输方式

3.1流式编码信息方式替换文本消息的传输

基于对标准文本信息流式化编码的设计，这里引入了Google的Protocol Buffers技术来将先前的文本测试数据序列化封装，通过设计流量测试用例程序来模拟流式文件的传输过程并于纯文本用例进行性能对比。

[import "common.header.proto"；

package amqp；

message test

{ required common.header header = 1； //信息头扩展，这里暂不定义

required int32 tid = 2； //信息编号

required int32 length = 3； //数据长度

required bool end_flag = 4； //是否结尾的标志位

repeated bytes data = 5； //数据段

}]

例1 amqp.test.proto文件实例

Protocol Buffers（Protobuf，官方译名“协议缓冲”）是Google公司开发的一种数据描述语言，类似于XML能够将结构化数据序列化，可用于数据存储、通信协议等方面。现阶段支持C++、JAVA、Python等三种编程语言。它的工作方式是通过编辑.proto文件来定义需要序列化信息的格式，每个buffer消息是一个逻辑记录，包括一系列的名称和值的组合甚至是架构子数据结构。Protobuf有如XML，不过它更小、更快、也更简单。你可以定义自己的数据结构，然后使用代码生成器生成的代码来读写这个数据结构。

针对数据的传输，编译Protobuf定义文件amqp.test.proto生成的对应数据类功能代码。其中对于少量数据情形，会将多个数据包通过一个数据结构（见例1. repeated声明段落）来进行预先并包。较大数据则是拆分为多个字节段，并提供了数据编号和数据长度等元信息，以保证数据校验。

根据该数据类提供的高效序列化和反序列化方法来对测试用例程序的消息发送和接收部分进行高层封装来提供编码支持。测试数据依然是以100k数据流量来分别用100～1000条数据发送，通过多次试验获得平均性能。结合标准用例来比较编码后的传输效率。

图6 数据传输对比

在检查效率提升的同时，选取较大数据源（100m）来测试传输方式的数据完整性，多次试验的结果依然是100%。

3.2 应用可行性

本文中解决的问题主要为长段数据的分包和解包问题，消息路由模型的选择和流式数据基础处理，是常规情况下消息中间件都需要处理在大数据流量情况下需要解决的主要问题。而在实际应用中，特别是互联网应用，需要考虑的问题会更多。比如由于网络问题而导致部分数据丢包、缺包甚至于断线重连的问题，这就需要对数据校验、请求应答和传输恢复等机制这些方面多加以考虑。再比如说服务器端的稳定性，由于AMQP对于数据的持久性还没有特别好的解决（具有“durable”设置但还远远不够），所以需要把大量的数据转存管理，并且一旦出现异常，对于数据的保存和恢复也应该要有相应的途径。

4 结束语

相对于HTTP、FTP和MMS这样的常用数据协议，作为新技术AMQP显得并不成熟并且认识度偏窄，但是它的前景已然可观，该协议具有较高的集成度，性能也不输（现今瓶颈都在硬件部分），设计规范，易于使用，部分实现已经成熟使用，发展也十分迅速，相信，在未来几年里，AMQP会得到更加广泛的应用。

参考文献：

[1] AMQP Working Group. Advanced message queuing protocol， 2010.http：//www.amqp.org.

[2] O'Hara， J. "Toward a commodity enterprise middleware".Acm Queue 5：48–55， 2007.doi： 10.1145/1255421.1255424.

[3] Vinoski， S. "Advanced Message Queuing Protocol". Internet Computing， IEEE， 10（6）： 87-89， 2006.doi：10.1109/MIC.2006.116.

[4] Appel， Stefan （TU Darmstadt， Germany）； Sachs， Kai； Buchmann， Alejandro. "Towards benchmarking of AMQP". Proceedings of the 4th ACM International Conference on Distributed Event-Based Systems， DEBS 2010： 9-100， 2010.doi：10.1145/1827418.1827438.

[5] Xiong， Xuandong； Fu， Jiandan. "Active Status Certificate Publish and Subscribe Based on AMQP". Computational and Information Sciences （ICCIS）， 2011 International Conference on： 725-728， Oct， 2011. doi： 10.1109/ICCIS.2011.63.