本文分享自华为云社区《物联网常见协议之 Amqp 协议及使用场景解析》，作者：张俭。
https://bbs.huaweicloud.com/blogs/397026

  引言

本文围绕 AMQP 协议，为大家详细解析 AMQP 协议、核心技术亮点、多协议之间的对比以及使用实践，并介绍华为云 IoT 通过 Amqp 协议如何为开发者和企业提供了更加灵活和高效的通信方式，使得物联网应用得以在各个领域得到更广泛的推广和应用。
AMQP 协议，全称为 Advanced Message Queuing Protocol。在 2006 年 6 月，由 Cisco、Redhat、iMatrix 等联合制定了 AMQP 的消息标准。
除了 AMQP 协议，还有一些其他协议如 Mqtt (Message Queuing Telemetry Transport)、Http、Kafka。每个协议的发明 / 出现都是为了解决特定的问题。没有最合适的协议，只有更合适的业务场景。在后面我们也会对这些协议进行简单的对比。
Amqp 历史上大概有如下四个版本，

  • Amqp 0-8: 发布于 2006 年
  • Amqp 0-9-1：发布于 2008 年，是 Amqp 0-8 的改进版，被广泛应用，如 rabbitmq、qpid 等
  • Amqp 0-10：发布于 2008 年，是 Amqp 0-9-1 的改进，未被广泛使用
  • Amqp 1.0：发布于 2011 年，是 Amqp 协议的下一代标准，与之前的版本不兼容，但提供了更强大的特性和更好的性能。目前也在华为云 IoT、Azure 中有应用起来。包括 rabbitmq、qpid 等也提供了对 Amqp1.0 版本协议的支持
我们也会主要讨论 Amqp 0-9-1 和 Amqp 1.0 这两个版本

  Amqp 0-9-1 协议简述
  核心概念

  • Virtual Host：简称 vhost，个人理解是 Amqp 协议上的多租，每个 vhost 具有自己的 Exchanges、Message Queues 等，互相不干扰。
  • Exchange: 从生产者应用程序中接收消息，并根据特定的情况（消息属性或内容），将这些消息路由到 “Message Queue” 中
  • Message Queue: 消息队列，存储消息，直到它被消费者应用程序安全地处理
  • Binding：指 Exchange 将何种类型的消息发送到 Queue 中，提供消息路由机制
Amqp 0-9-1 协议是一个 多链路、协商的、异步、安全、可移植、高效的协议。Amqp 协议通常分为两层:
  
  1. +------------------Functional Layer----------------+
  2. | Basic Transactions Exchanges Message queues     |
  3. +--------------------------------------------------+
  4. +------------------Transport Layer-----------------+
  5. | Framing Content Data representation             |
  6. | Error handling Heart-beating Channels |
  7. +--------------------------------------------------+

此外，由于 Amqp 协议的 message queue 支持许多特性：私有或共享、持久化或临时等等。根据不同的属性设定，我们可将 AMQP 用于许多应用场景，例如

  • 消息中间件使用：共享的存储转发队列，保存消息并交给多个消费者消费
  • RPC 使用：通过将队列设定为临时的，带有 IP 地址的，来模拟 RPC 接口
  
Amqp 0-9-1 生产时序图
    v2-dfca5013de741dd43b4e6531cfe54264_720w.jpg

  Amqp 0-9-1 消费时序图
    v2-d639050953f940b6271fcb936ec6787a_720w.jpg

  Amqp 0-9-1 协议帧及数据类型
Amqp 0-9-1 的协议帧由 FrameHeader、Payload、FrameEnd 组成

  • Integers 整数（1 到 8 个字节）：用于表示大小、数量、限制等。整数总是无符号的，
  • Bits 位：用于表示开 / 关值，一个八位字节。
  • Short strings 短字符串：用于存储短文本属性。短字符串长度限制为 255 个八位字节。
  • Long strings 长字符串：用于存储二进制数据块。
  • Field tables 字段表：存储名称 - 值对。字段值可以是字符串、整数等类型。

  Amqp 1-0 协议
  与 Amqp 0-9-1 的差异
协议设计层面：

  • AMQP 0-9-1：此版本的 AMQP 主要针对代理的设计，涵盖了消息传递模型、代理行为和交互模式。0-9-1 版本的协议与代理的实现紧密耦合。
  • AMQP 1.0：此版本的 AMQP 更注重基于互操作性的通信协议，不依赖于特定的代理实现。AMQP 1.0 关注点在于在发送者和接收者之间传输消息，而不是代理的内部行为。
  • 比如像”Queue Declare”、“Queue Delete”、“Queue Query” 这些在 Amqp 0-9-1 支持的命令，在 Amqp1.0 中都被移除，并假设这些功能会在更高层（broker）参加。
对称层面：

  • Amqp 0-9-1 是一个典型的客户端 / 服务器通信协议。
  • Amqp 1.0 则是一个对称的协议，任何一端都可以注册为 sender 或是 receiver，并且从如下 Amqp 0.9.1 和 1.0 之间的时序图对比也可以看出来。Amqp 1.0 是完全双工的协议。从某种程度或者说网络编程的角度来说，实现的难度更大。
  
Amqp 1-0 鉴权时序图
    v2-89a38b30f9e64b46a78e4c7ecb7efd7b_720w.jpg

  Amqp 1-0 生产时序图
    v2-0023ddc8664a7b7f00dc36ce955627e4_720w.jpg

  Amqp 1-0 消费时序图
    v2-a6442bcaec5647aa05a50072fca77ca7_720w.jpg

  Amqp 1.0 协议帧介绍
    v2-d7c3e9a285d413d5b4036134cc619346_720w.jpg
Amqp1.0 的协议帧由 FrameHeader、ExtendedHeader、FrameBody 组成。

  • FrameHeader 8 个字节大小，包含长度、类型信息等
  • Extended header 可变宽度区域
  • FrameBody 是一个可变宽度的字节序列，其格式取决于帧类型
  
FrameHeader 介绍
    v2-ccf916bd0fb0198a2665c834b6cbbbdf_720w.jpg

  • Size: FrameHeader 的第 0~3 个字节包含帧大小。无符号的 32 位整数，为 FrameHeader、ExtendedHeader、FrameBody 的总和大小。如果大小小于 8 字节，则格式错误
  • DOFF: FrameHeader 的第 4 个字节，这表示帧内 Body 的位置。
  • Type: FrameHeader 的第 5 个字节，类型代码表示帧的格式和目的。根据帧的类型，帧头中的后续字节可能会被不同地解释。类型代码 0x00 表示该帧是 AMQP 帧。类型代码 0x01 表示该帧是 SASL 帧等。
  
Amqp 帧介绍
Amqp 帧类型代码为 0x00。对于 Amqp 帧来说，FrameHeader 的第 6 字节和第 7 字节表示 channel 的编号。Frame Body 被定义为一个 performative 后跟一个不透明的 payload。表现形式必须是第 open、begin、attach、flow、transfer、disposition、detach、end、close 中定义的一个，并在 AMQP 类型系统中编码为描述的类型。帧体中剩余的字节构成了该帧的 payload。payload 的存在和格式由给定表现形式的语义定义。

  SASL 帧介绍
Sasl 帧类型代码为 0x01。FrameHeader 中的第 6 和第 7 字节应该被忽略。也不存在扩展头。所以 DOFF 固定位 0x02。

  与其他消息通信协议间的对比
  Amqp 与 Mqtt 的对比
Amqp 和 Mqtt 都是应用层的消息传递协议，mqtt 更加轻量，相对来说概念不如 amqp 那么丰富，同时 mqtt 头部消息更加短小。更加适用于低带宽、功耗较低的物联网设备

  Amqp 与 Kafka 协议的对比
AMQP 是一种非常灵活的协议，可以用于各种类型的消息传递场景，包括点对点和发布 - 订阅模型。Kafka 则专注于高吞吐量的流式处理，适用于数据管道和流式处理等场景。
Kafka 的设计旨在提供高吞吐量和低延迟。AMQP 的性能因实现和使用情况而异，但在大多数情况下，它的性能不如 Kafka。
Kafka 拥有强大的生态系统，包括流处理、数据湖、消息队列等多个应用场景。AMQP 也有相应的生态系统和工具，但相对来说要小得多。
总得来说，尽管 kafka 存在性能上的优势，但 kafka broker 很难对外暴露。相较于 kafka 这种私有消息中间件协议，Amqp 足够标准，更适合各种异构系统的对接。

  AMQP 协议相关的开源项目
  rabbitmq
提到 AMQP，就不得不提 rabbitmq。RabbitMQ 是一个开源的消息代理和队列服务器，用于通过高级消息队列协议（AMQP）在分布式系统中实现消息传递。RabbitMQ 提供了一个可靠、高性能、可扩展和易于使用的消息传递平台，支持多种编程语言和平台。它最初是用 Erlang 语言编写的，因此具有良好的并发性能和容错能力。
所谓成也 erlang，败也 erlang，由于 erlang 语言生态的问题，有能力深入维护 Rabbitmq 的人员并不是很多，也是 rabbitmq 越来越不流行的一个原因。

  Qpid
Apache Qpid（Quick Platform for Interactive Distributed Messaging）是一个开源的消息传递系统，它实现了高级消息队列协议（Advanced Message Queuing Protocol，AMQP）的多种版本。AMQP 是一种开放标准的应用层协议，用于消息传递的中间件，它可以实现跨平台、跨语言的消息通信。Qpid 项目的主要目标是提供一个可靠、可扩展和高性能的消息传递平台，帮助开发者更容易地构建分布式系统。主要的组件有

  • Qpid Broker：一个高性能、可扩展的 AMQP 消息代理，支持持久化、事务和安全认证等特性。Qpid Broker 提供了 Java 和 C++ 两种实现。
  • Qpid Proton：一个轻量级的 AMQP 库，旨在为各种编程语言提供高性能的 AMQP 实现，提供了 C 和 java 的默认实现。此外 Proton 还提供了其他编程语言如 python 的绑定。
但 Qpid 总得来说，比较重型，如果仅仅是想在原有的消息组件，如 kafka/pulsar 外面叠加一层 Amqp 可访问的能力，我相信 proton 是更好的选择。

  Vertx-proton
Vert.x Proton 的目标是结合 Vert.x 的响应式编程模型和 Qpid Proton 的 AMQP 支持，以简化构建高性能、可扩展的、基于 AMQP 的分布式应用程序。Vert.x Proton 提供了一套简洁、易用的 API，可以让开发者在 Vert.x 应用程序中轻松地实现 AMQP 通信。

  华为云 IoT 对 AMQP 的支持
在最初阶段华为云 IoTDA 主要支持 HTTP 协议，尽管这种方式已经能满足许多需求，但随着物联网技术的普及和发展，用户对于更加灵活和高效的通信方式的需求逐渐增强，华为云 IoTDA 逐渐丰富协议库，当前支持 60 + 协议接入，为开发者和企业提供更加完善的解决方案。
在 IoT 应用对接场景中，华为云 IoT 现已新增了对 AMQP 的支持，与 HTTP 协议相比，AMQP 协议具有以下优势

  • 无需 HTTP 服务器：AMQP 协议无需开发者搭建 HTTP 服务器，降低了项目成本，简化了系统架构。可以部署在各种类型的设备，包括手机、平板、智能家居设备等，进一步拓宽了物联网应用的领域。
  • 低延迟、高效率：AMQP 协议采用二进制传输，降低了数据传输所需的带宽，提高了传输速度，降低了延迟。
  • 强大的消息队列功能：AMQP 协议具有优秀的消息队列功能，支持点对点和发布订阅模式，确保消息的可靠传输和顺序处理。
通过支持 AMQP 协议，华为云 IoT 为开发者和企业提供了更加灵活和高效的通信方式，使得物联网应用得以在各个领域得到更广泛的推广和应用。

  Amqp 实战：使用 qpid-proton python 消费华为云 IoTDA 的 Amqp 消息
首先通过 pip 安装依赖包
  
  1. pip install python-qpid-proton

最简单的消费者 demo, consumer.py 如下
  
  1. import sys
  2. from proton.handlers import MessagingHandler
  3. from proton.reactor import Container
  4. class AMQPConsumer(MessagingHandler):
  5. def __init__(self, server_url, target_address):
  6. super(AMQPConsumer, self).__init__()
  7. self.server_url = server_url
  8. self.target_address = target_address
  9. def on_start(self, event):
  10.         conn = event.container.connect(self.server_url)
  11. event.container.create_receiver(conn, self.target_address)
  12. def on_message(self, event):
  13. print(f"Received message: {event.message.body}")
  14. event.connection.close()
  15. if __name__ == "__main__":
  16. server_url = "amqp://localhost:5672"
  17. target_address = "example_queue"
  18. try:
  19. Container(AMQPConsumer(server_url, target_address)).run()
  20. except KeyboardInterrupt:
  21. sys.exit(0)

我们可以使用这个 producer.py 验证 consumer.py 可用
  
  1. import sys
  2. from proton import Message
  3. from proton.handlers import MessagingHandler
  4. from proton.reactor import Container
  5. class AMQPProducer(MessagingHandler):
  6. def __init__(self, server_url, target_address, message_body):
  7. super(AMQPProducer, self).__init__()
  8. self.server_url = server_url
  9. self.target_address = target_address
  10. self.message_body = message_body
  11. def on_start(self, event):
  12.         conn = event.container.connect(self.server_url)
  13. self.sender = event.container.create_sender(conn, self.target_address)
  14. def on_sendable(self, event):
  15.         message = Message(body=self.message_body)
  16. event.sender.send(message)
  17. print(f"Sent message: {message.body}")
  18. event.connection.close()
  19. if __name__ == "__main__":
  20. server_url = "amqp://localhost:5672"
  21. target_address = "example_queue"
  22. message_body = "Hello, AMQP 1.0!"
  23. try:
  24. Container(AMQPProducer(server_url, target_address, message_body)).run()
  25. except KeyboardInterrupt:
  26. sys.exit(0)

为了能连接上华为云 IoTDA 的 Amqp 接入点，我们还需要给 consumer.py 配置用户名、密码。如果您对 TLS 认证服务端还有诉求，则还可以配置上证书信息。如下为样例代码，具体连接的信息、凭据如何获得可参考: https://support.huaweicloud.com/devg-iothub/iot_01_00100_2.html。注意，url 也从 amqp 修改为了 amqps
  
  1. import sys
  2. from proton import Message, SSLDomain
  3. from proton.handlers import MessagingHandler
  4. from proton.reactor import Container
  5. class AMQPConsumer(MessagingHandler):
  6. def __init__(self, server_url, target_address, username, password, cert_file, key_file):
  7. super(AMQPConsumer, self).__init__()
  8. self.server_url = server_url
  9. self.target_address = target_address
  10. self.username = username
  11. self.password = password
  12. self.cert_file = cert_file
  13. self.key_file = key_file
  14. def on_start(self, event):
  15. ssl_domain = SSLDomain(mode=SSLDomain.MODE_CLIENT)
  16. ssl_domain.set_credentials(self.cert_file, self.key_file, None)
  17.         conn = event.container.connect(self.server_url, user=self.username, password=self.password, ssl_domain=ssl_domain)
  18. event.container.create_receiver(conn, self.target_address)
  19. def on_message(self, event):
  20. print(f"Received message: {event.message.body}")
  21. event.connection.close()
  22. if __name__ == "__main__":
  23. server_url = "amqps://localhost:5671" # 注意 'amqps'，它表示使用 SSL/TLS 连接
  24. target_address = "example_queue"
  25.     username = "your_username"
  26.     password = "your_password"
  27. cert_file = "path/to/your/certificate.pem"
  28. key_file = "path/to/your/private_key.pem"
  29. try:
  30. Container(AMQPConsumer(server_url, target_address, username, password, cert_file, key_file)).run()
  31. except KeyboardInterrupt:
  32. sys.exit(0)

该样例代码已上传到 gitee

  总结与展望
总体来说，AMQP 作为一种应用层协议，在消息传递和异构系统之间的通信方面提供了非常灵活和可靠的解决方案。与其他消息传递协议相比，AMQP 具有丰富的功能和灵活的设计，适用于各种类型的消息传递场景。
在使用 AMQP 时，我们可以选择现有的开源实现，如 RabbitMQ、Qpid 等，也可以自行实现 AMQP 的相关组件。通过这些实现，我们可以轻松地在不同的应用程序、语言和平台之间进行消息传递，并实现可靠、高效、安全的通信。
随着物联网、云计算和大数据等技术的发展，AMQP 的应用场景越来越广泛，比如在 IoT 设备、大数据流处理、分布式系统等方面都得到了广泛应用。未来，AMQP 将继续发挥重要作用，推动各种异构系统之间的互联互通，带来更加便捷和高效的消息传递体验。